DEV Community: suzuki-navi

How to Extract Every Nth Line from a Text File

suzuki-navi — Wed, 08 Mar 2023 12:44:33 +0000

To output every fifth line starting from the first line, sixth line, eleventh line, and so on,

$ sed -ne 1~5p < data.txt
$ awk NR%5==1 < data.txt
$ perl -nle '$.%5==1 && print' < data.txt
$ ruby -nle '$.%5==1 && print' < data.txt

To output every fifth line starting from the fifth line, tenth line, fifteenth line, and so on,

$ sed -ne 0~5p < data.txt
$ awk NR%5==0 < data.txt
$ perl -nle '$.%5==0 && print' < data.txt
$ ruby -nle '$.%5==0 && print' < data.txt

At first glance, Perl and Ruby may appear to be the same.

Script to Check the Timeout Value and Recent Execution Time of a Glue Job

suzuki-navi — Thu, 09 Feb 2023 14:01:30 +0000

I wrote a script that calculates the minimum, average, and maximum time taken from the recent execution history of the Glue Job and displays it along with the Timeout value set for the Glue Job.

It is displayed in this way.

name                                                                   timeout     min    avg    max
gluejob-test2                                                           172800      36     41     44
gluejob-test2-01                                                        172800      52     54     57
longgluejob-sample1                                                     172800    2621   3071   3725
longgluejob-sample2                                                     172800   20837  22760  27465

The unit of time is in seconds.

The source code is like this. It is written in Python.

import sys
import boto3

def main():
    profile = sys.argv[1]
    list_jobs(profile)

def list_jobs(profile):
    session = boto3.session.Session(profile_name=profile)
    glue_client = session.client("glue")
    jobs = fetch_glue_job_list(glue_client)
    print("name                                                                   timeout     min    avg    max")
    for job in jobs:
        job_name = job["Name"]
        timeout = job["Timeout"] * 60 # the values returned by the API are in minutes.
        history = fetch_glue_job_history(glue_client, job_name)

        job_name_str = job_name.ljust(70)
        timeout_str = str(timeout).rjust(6)
        history_str = " ".join([str(h).rjust(6) for h in history])
        print(f"{job_name_str} {timeout_str} {history_str}")

def fetch_glue_job_list(glue_client):
    jobs = []
    res = glue_client.get_jobs()
    while True:
        for elem in res["Jobs"]:
            jobs.append(elem)
        if "NextToken" not in res:
            break
        res = glue_client.get_jobs(NextToken=res["NextToken"])
    return jobs

def fetch_glue_job_history(glue_client, job_name):
    MAX_COUNT = 100 # retrieve 100 as the recent execution history
    history = []
    res = glue_client.get_job_runs(JobName=job_name)
    while True:
        for run in res["JobRuns"]:
            if run["JobRunState"] == "SUCCEEDED":
                history.append(run["ExecutionTime"])
                if len(history) >= MAX_COUNT:
                    break
        if len(history) >= MAX_COUNT:
            break
        if "NextToken" not in res:
            break
        res = glue_client.get_job_runs(JobName=job_name, NextToken=res["NextToken"])

    if history:
        avg_value = int(sum(history) / len(history))
        max_value = max(history)
        min_value = min(history)
    else:
        avg_value = 0
        max_value = 0
        min_value = 0
    return [min_value, avg_value, max_value]

main()

Add a WHERE clause to the JDBC data source of a Glue Job

suzuki-navi — Tue, 07 Feb 2023 01:47:33 +0000

This is a method for processing only a portion of data from a data source in a Glue Job by adding a WHERE clause to the access with JDBC.

It seams impossible to do with DynamicFrame, so I use DataFrame.

When connecting with DataFrame, it seams necessary to explicitly specify the connection information with JDBC. I use glueContext.extract_jdbc_conf to retrieve the information from the Glue Connection.

from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.dynamicframe import DynamicFrame

sparcContext = SparkContext()
glueContext = GlueContext(sparcContext)
logger = glueContext.get_logger()
sparkSession = glueContext.spark_session

CONNECTION_NAME = "test-connection" # Name of Glue Connection
DB_NAME = "testdb" # Name of the PostgreSQL database to connect to

# Retrieve the information from the Glue Connection
jdbc_conf = glueContext.extract_jdbc_conf(connection_name=CONNECTION_NAME)
logger.info(jdbc_conf)
# Outputs as follows:
# {'enforceSSL': 'false', 'skipCustomJDBCCertValidation': 'false', 'url': 'jdbc:postgresql://hogehoge.xxxxxxxxxxxx.ap-northeast-1.rds.amazonaws.com:5432', 'customJDBCCertString': '', 'user': 'postgres', 'customJDBCCert': '', 'password': 'xxxxxxxxxxxx', 'vendor': 'postgresql'}

# The string in jdbc_conf["url"] is in the format
# jdbc:postgresql://HOST:5432
# and missing the database name at the end.
# When passed to DynamicFrame.fromDF, adding the database name is necessary.
jdbc_url = jdbc_conf["url"] + "/" + DB_NAME

# The SQL is written as follows:
query = "(SELECT * FROM public.testtable WHERE id <= 100) t"

dyf0 = DynamicFrame.fromDF(sparkSession.read.format("jdbc").
    option("url", jdbc_url).
    option("driver", "org.postgresql.Driver").
    option("user", jdbc_conf["user"]).
    option("password", jdbc_conf["password"]).
    option("dbtable", query).load(), glueContext, "dyf0")

# Do something

When this Glue Job is executed, the following query is issued to PostgreSQL.

SELECT "id","col1","col2","col3" FROM (SELECT * FROM public.testtable WHERE id <= 100) t