Informatica - grep command in Unix/Linux

The grep filter searches a file for a particular pattern of characters, and displays all lines that contain that pattern. The pattern that is searched in the file is referred to as the regular expression (grep stands for globally search for regular expression and print out). 

Syntax: 
   grep [options] pattern [files]

Options Description
-c : This prints only a count of the lines that match a pattern
-h : Display the matched lines, but do not display the filenames.
-i : Ignores, case for matching
-l : Displays list of a filenames only.
-n : Display the matched lines and their line numbers.
-v : This prints out all the lines that do not matches the pattern
-e exp : Specifies expression with this option. Can use multiple times.
-f file : Takes patterns from file, one per line.
-E : Treats pattern as an extended regular expression (ERE)
-w : Match whole word
-o : Print only the matched parts of a matching line,
 with each such part on a separate output line.

-A n : Prints searched line and nlines after the result.
-B n : Prints searched line and n line before the result.
-C n : Prints searched line and n lines after before the result.

e.g.

find /Scripts/ -name \*.\*sh -exec grep -ils "table_name" {} \; 2>/dev/null | sort | uniq

SQL Server - List all table_names, row_count, usage, view_names, view_definitions from database

--Table

SELECT

s.Name AS SchemaName,

t.Name AS TableName,

p.rows AS RowCounts,

CAST(ROUND((SUM(a.used_pages) / 128.00), 2) AS NUMERIC(36, 2)) AS Used_MB,

CAST(ROUND((SUM(a.total_pages) - SUM(a.used_pages)) / 128.00, 2) AS NUMERIC(36, 2)) AS Unused_MB,

CAST(ROUND((SUM(a.total_pages) / 128.00), 2) AS NUMERIC(36, 2)) AS Total_MB

FROM sys.tables t

INNER JOIN sys.indexes i ON t.OBJECT_ID = i.object_id

INNER JOIN sys.partitions p ON i.object_id = p.OBJECT_ID AND i.index_id = p.index_id

INNER JOIN sys.allocation_units a ON p.partition_id = a.container_id

INNER JOIN sys.schemas s ON t.schema_id = s.schema_id

GROUP BY t.Name, s.Name, p.Rows

ORDER BY 3 desc, 4

You can get data like this:

--View

SELECT v.name as VIEW_NAME, v.create_date , Replace(Replace(SM.definition,char(10),''),char(13),'') as view_definition

FROM sys.views V

     INNER JOIN  sys.sql_modules SM ON V.object_id  = SM.object_id 

     INNER JOIN sys.Objects SO ON SM.Object_id = SO.Object_id

WHERE SO.type = 'v'

Databricks SQL - extract data from nested JSON text

 

%sql

select distinct left(v2.inner_key, locate('_', v2.inner_key) - 1 ) as some_alias

from db_name.tbl_name 

     LATERAL VIEW json_tuple(tbl_name.nested_json_col_name, 'outter_key') v1 

     as outer_key

     LATERAL VIEW json_tuple(v1.outer_key, 'inner_key') v2

     as inner_key

Datadog event - function to capture notebook exception

This function can be used at any "except Exception" like below. e.g. 

try:
 ## any code logic
  
except Exception as e:   
  context_str = dbutils.notebook.entry_point.getDbutils().notebook().getContext().toJson()
  DatabricksNBRuntimeError(context_str)

Functions are:

from datadog import initialize, api

api_key = dbutils.secrets.get(scope="datadog", key="datadog_api_key")
app_key = dbutils.secrets.get(scope="datadog", key="datadog_app_key")

options = {
    "api_key": f"{api_key}",
    "app_key": f"{app_key}"
}

initialize(**options)

import sys
import requests
import traceback
import json

def datadog_event(notebook_path, job_name, user):
    exc_type, exc_value, exc_tb = sys.exc_info()
    traceback_error=str(traceback.format_exception(exc_type, exc_value, exc_tb))
    error_case_list = [find_between(traceback_error,"Exception","\n\tat"), find_between(traceback_error,"Error","\n\tat"), traceback_error]
    error_message = next(error for error in error_case_list if error != '')
      
    error_dict = {"notebook_metadata": {"job_name":job_name, "notebook_path": notebook_path,"user": user},  "error": error_message}
    error_message = json.dumps(error_dict)
    
    title = datadog_identify_string
    if job_name:
      title = datadog_identify_string + ': job( ' + job_name + ' )'
      
    text = error_message

    print(title, text)
    api.Event.create(title=title, text=text, tags=datadog_tags, priority='Low')   
def exit_notebook(job_name, user):
  
  error_case_list = [find_between(str(e),"Exception","\n\tat"), find_between(str(e),"Error","\n\tat"), str(e)[:200]]
  error_message = next(error for error in error_case_list if error != '')
  output = {"status": "FAIL", "notebook_metadata": {"job_name":job_name, "user":user}, "message": error_message}
  
  dbutils.notebook.exit(json.dumps(output))
class DatabricksNBRuntimeError(Exception):
 
  def __init__(self, context_str, **kwargs):
       
    self.strerror = args
    self.args = args
        
    notebook_metadata = json.loads(context_str)
    
    job_name = None
    user = None
    notebook_path = notebook_metadata['extraContext']['notebook_path']
    if 'jobName' in notebook_metadata['tags'].keys():
      job_name = notebook_metadata['tags']['jobName']
    if 'user' in notebook_metadata['tags'].keys():
      user = notebook_metadata['tags']['user']
      
    datadog_event(notebook_path, job_name, user) 
    exit_notebook(job_name, user)

Argo workflow - Prometheus metrics

https://argoproj.github.io/argo-workflows/metrics/#metric-spec

prometheus:
  # Metric name (will be prepended with "argo_workflows_").
  # Metric names can only contain alphanumeric characters, _, and :.
  - name: metric_failed_with_output_result
    # Metric definitions must include a name and a help doc string.
    # (this is a Prometheus requirement)Metrics with the
    # same name must always use the same exact help string,
    # having different metrics with the same name,
    # but with a different help string will cause an error.
    help: "test run"
    # Labels are optional. Avoid cardinality explosion.
    labels:
      - key: workflow_name
        value: "{{ workflow.name }}"
      - key: workflow_namespace
        value: "{{ workflow.namespace }}"
      - key: status
        # different workflows may fail with the same error, 
        # so add unique workflow as prefix
        value: "{{ workflow.name }}_{{ outputs.result }}"
    # Emit the metric conditionally. Works the same as normal "when"
    when: "{{ status }} == Failed"
    # The metric type. Available are "gauge", "histogram", and "counter".
    gauge:
      # This increments the counter by 1
      value: "{{duration}}"

AWS - Restore RDS

 

Step 0: Update RDS backup file to s3

aws configure set default.s3.max_concurrent_requests 45
aws configure set default.s3.multipart_threshold 128MB
aws configure set default.s3.multipart_chunksize 64MB
aws configure set default.s3.max_bandwidth 150MB/s
aws s3 cp "C:\Backup\xxxxxx.bak" s3://xxx/xxx/

Step 1: Create RDS instance

aws rds create-db-instance \
--engine sqlserver-ee \
--license-model license-included  \
--engine-version 11.00.7493.4.v1 \
--db-instance-identifier XXXXXX \
--storage-type gp2 \
--db-instance-class db.m5.xlarge \
--max-allocated-storage 4000 \
--allocated-storage 3000 \
--master-username XXXXXX \
--master-user-password XXXXXX \
--db-subnet-group XXXXXX \
--vpc-security-group-ids XXXXXX \
--port 1433 \
--backup-retention-period 0 \
--option-group-name XXXXXX \
--publicly-accessible

Step 2: Create RDS Option Groups

aws rds create-option-group \\
          --option-group-name XXXXX \\
          --engine-name sqlserver-ee \\
          --major-engine-version 11.00.7493.4.v1 \\
          --option-group-description "restore sql server bak file"

Step 3: Add Option to Option Groups to add SQL DB Backup option from S3

aws rds add-option-to-option-group \\
          --option-group-name XXXXXX \\
          --options "OptionName=SQLSERVER_BACKUP_RESTORE,OptionSettings=
                     [{Name=IAM_ROLE_ARN,Value=arn:aws:iam::XXXXXX}]" \\
          --apply-immediately

Step 5: Connect to this RDS instance from local SQL Server DB

The proper connection details for DBeaver for me were:

• Server Host: Endpoint address
• Port: 1433
• Database: DB name
• User name: Database master username
• Password: Database master password

Step 6: Delete RDS instance

-- aws rds delete-db-instance \
--db-instance-identifier xxxxxx \
--skip-final-snapshot

SQL Server Restore backup:

1. Restore database: exec msdb.dbo.rds_restore_database @restore_db_name=‘databasename’ , @s3_arn_to_restore_from=‘arn:aws:s3:::s3bucketname/databasename.bak’
2. Check status: exec msdb.dbo.rds_task_status;

Databricks Spark -Cheatsheet Summary

 

Import

dbutils.library.installPyPI("xlsxwriter")

from pyspark.sql.functions import *

from delta.tables import DeltaTable

from pyspark.sql import DataFrame

import pyspark.sql.functions as F

from pyspark.sql import types as T

from pyspark.sql.functions import col,current_timestamp,from_json,

from_unixtime,lag,lead,lit,mean,stddev,max,regexp_replace

from pyspark.sql.session import SparkSession

from pyspark.sql.streaming import DataStreamWriter

from pyspark.sql.window import Window

Files

from delta.tables import *

spark.conf.set("spark.databricks.delta.retentionDurationCheck.enabled", "false")  

#bypass retention duration so that can vacuum

deltaTable = DeltaTable.forPath(spark, deltaPath)

deltaTable.vacuum(0)

dbutils.fs.cp(tempfile,fileprefix+".csv")

dbutils.fs.put(result_path, sms_m, overwrite=True)

dbutils.fs.mv("file:/databricks/driver/health_tracker_data_2020_1.json", 

health_tracker + "raw/health_tracker_data_2020_1.json")

dbutils.fs.rm(health_tracker + "processed", recurse=True)

%fs rm -r /dbacademy/DLRS/healthtracker/gold/health_tracker_user_analytics

display(dbutils.fs.ls(rawPath))

print(dbutils.fs.head(suppliersPath))

# To check the file you need(file_2020_1) is in the directory(rawPath)

assert file_2020_1 in [item.name for item in dbutils.fs.ls(rawPath)], 

"File not present in Raw Path"

# Mount using the s3a prefix

dbutils.fs.mount(bucketURL, MOUNTPOINT)

# Unmount directory if previously mounted

if MOUNTPOINT in [mnt.mountPoint for mnt in dbutils.fs.mounts()]:

  dbutils.fs.unmount("/mnt/s3demo")

Load

1. General

df = spark.read.load("examples/src/main/resources/users.json", format="json")

df = spark.read.load("examples/src/main/resources/users.csv", format="csv", sep=";", inferSchema="true", header="true")

df = spark.read.load("examples/src/main/resources/users.parquet")

# pathGlobFilter is used to only include files with file names matching the pattern

df = spark.read.load("examples/src/main/resources/dir1", format="parquet", pathGlobFilter="*.parquet")

# recursiveFileLookup is used to recursively load files and it disables partition inferring. Its default value is false. 

# If data source explicitly specifies the partitionSpec when recursiveFileLookup is true, exception will be thrown. 

# To load all files recursively, you can use:

df = spark.read.format("parquet").option("recursiveFileLookup", "true").load("examples/src/main/resources/dir1")

df = spark.read.format("delta").option("versionAsOf", 0).load("examples/src/main/resources/dir1")

df = spark.read.format("xml").option("rootTag", "orders").option("rowTag", "purchase_item").option("inferSchema", "true").load(purchaseOrdersPath)

df = spark.sql("SELECT * FROM parquet.`examples/src/main/resources/users.parquet`")

df = spark.read.table("health_tracker_processed")

df = spark.read.option("mergeSchema", "true").parquet("examples/src/main/resources/users.parquet")

df = spark.read.option("header", "true").option("inferSchema", "true").option("nullValue", "nan").csv("examples/src/main/resources/users.csv")

df = spark.read.option("inferSchema", "true")json("examples/src/main/resources/users.json")

2. Streaming

df_stream = spark.readStream.format("text").option("maxFilesPerTrigger", 1).load(rawPath)

df_stream = spark.readStream.schema(jsonSchema).option("maxFilesPerTrigger", 1).json(inputPath)\

                            .withColumn("INPUT_FILE_NAME", input_file_name()) \ #maintain file path

                            .withColumn("PROCESSED_TIME", current_timestamp()) \#add a processing timestamp at the time of processing

                            .withWatermark("PROCESSED_TIME", "1 minute") #optional: window for out of order data

3. SQL

%sql CREATE DATABASE IF NOT EXISTS dbacademy USE dbacademy

%sql SET spark.sql.shuffle.partitions=8

--MSCK REPAIR TABLE recovers all the partitions in the directory of a table and updates the Hive metastore

%sql MSCK REPAIR TABLE health_tracker_processed

%sql DESCRIBE DETAIL health_tracker_processed

%sql DESCRIBE HISTORY health_tracker_silver

%sql DESCRIBE EXTENDED health_tracker_silver

%sql CONVERT TO DELTA parquet.`/dbacademy/DLRS/healthtracker/silver` PARTITIONED BY (p_device_id double)

%sql OPTIMIZE sc_raw_intake.raw_wims_events

CACHE TABLE device_data

%sql 

SELECT  dc_id,device_type, temps,TRANSFORM (temps, t -> ((t * 9) div 5) + 32 ) AS `temps_F`

FROM device_data;

%sql

SELECT dc_id, device_type, co2_level, REDUCE(co2_level, 0, (c, acc) -> c + acc, acc ->(acc div size(co2_level))) as average_co2_level

  FROM device_data

  SORT BY average_co2_level DESC;

%sql 

CREATE OR REPLACE TEMPORARY VIEW broken_readings

AS (

  SELECT COUNT(*) as broken_readings_count, dte FROM health_tracker_silver

  WHERE heartrate < 0

  GROUP BY dte

  ORDER BY dte

)

%sql

INSERT INTO health_tracker_silver

SELECT name,

       heartrate,

       CAST(FROM_UNIXTIME(time) AS TIMESTAMP) AS time,

       CAST(FROM_UNIXTIME(time) AS DATE) AS dte,

       device_id as p_device_id

FROM health_tracker_data_2020_02

%sql

DROP TABLE IF EXISTS health_tracker_silver;               -- ensures that if we run this again, it won't fail                                                        

CREATE TABLE health_tracker_silver                        

USING PARQUET                                             

PARTITIONED BY (p_device_id)                              -- column used to partition the data

LOCATION "/dbacademy/DLRS/healthtracker/silver"           -- location where the parquet files will be saved

AS (                                                      

  SELECT name,                                            -- query used to transform the raw data

         heartrate,                                       

         CAST(FROM_UNIXTIME(time) AS TIMESTAMP) AS time,  

         CAST(FROM_UNIXTIME(time) AS DATE) AS dte,        

         device_id AS p_device_id                         

  FROM health_tracker_data_2020_01   

)

spark.sql(

    """

DROP TABLE IF EXISTS xxx.xxxx

"""

)

spark.sql(

    f"""

CREATE TABLE xxx.xxx

USING DELTA

LOCATION "{destination_path}"

"""

)

spark.sql(f"CREATE DATABASE IF NOT EXISTS {cleaned_username}")

spark.sql(f"USE {cleaned_username}")

df.write.format("delta").mode("overwrite").saveAsTable("deltaReview")

Transform

1. select

df.select('xxxx.*')

df.select('operator').distinct()

# dteWindow = Window.partitionBy("p_device_id").orderBy("dte")

df.select(col("heartrate"), col("dte"),lag(col("heartrate")).over(dteWindow).alias("prev_amt"),lead(col("heartrate")).over(dteWindow).alias("next_amt"))

df.select(from_json(col("value"), json_schema_raw).alias("outer_json"))

df.select(lit("Tibco").alias("datasource"), current_timestamp().alias("ingesttime"), "value", current_timestamp().cast("date").alias("p_ingestdate"))

df.select("dte", "time", "device_type", "heartrate", lit(None).alias("name"), "p_device_id")

df.join(productsDF, on="product_name", how="inner")

df.join(df_planned, (df_actual.ITM_ID == df_planned.Item_code) & (df_actual.p_cre8date == df_planned.Date), how='left') 

df.createOrReplaceTempView("updates")

sql_query = "select count(*) as {}_count from {} where cast({} as date) = {}".format(stage, stage, ts_col, ts_date) 

spark.sql(sql_query).show()

df.selectExpr(['*',  'cast(CRE8_TS as timestamp) as NEW_CRE8_TS','cast(CRE8_TS as date) as CRE8_DT'])

2. filter

df.withColumn('dummy_var',F.rand())

df.withColumn('p_cre8date', F.col('CRE8_TS').cast("date"))

df.withColumn("banned_supplier", col("supplier").isin(supplierBlacklist))

df.withColumn('avg_weekly_unts',F.avg(F.col('wkly_unts')).over(prod_str_window))

df.withColumn("warehouse_alert", when(col("product_quantity") == 0, "red").otherwise("green"))

df.withColumn(col, F.when((F.lower(col) == un_list[0]) | (F.lower(col) == un_list[1]), 'undetermined').otherwise(F.lower(col)))

df.withColumn("row_num", row_number().over(Window.partitionBy([focus_term, gb_term]).orderBy(desc('year-month'), desc('value'))))

df.withColumn("price", translate(col("price"), "$,", "").cast("double"))

df.where(col("heartrate") < 0)

df.where(col("p_device_id").isin([3,4])

df.filter((col("ShipDateKey") > 20031231) & (col("ShipDateKey") <= 20041231))

df.filter((col("missing_price") == True) | (col("banned_supplier") == True))        

df.filter((F.col(focus_term).isNull()==False) | (F.col(gb_term).isNull()==False))

3. clean data

df.withColumnRenamed("TAX_ID", "tax_id")

df.withColumn('FW', F.coalesce(F.col('FW'), F.lit('unknown')))

df.dropDuplicates(["value"])

df.drop("product_quantity")

df.dropna(subset=(focus_term, gb_term))

4. aggregations

df.count()

df.orderBy(desc("load_timestamp"))

df.orderBy('ApproxWeight', ascending = False)

df.groupBy(list_g).agg(count(date_type).alias('value'))

#because lazy execution, you can cast as int after sum even if it is string type

df.groupby('DTA_AREA_ID').agg(F.sum(F.col('SALS_QTY')).cast('int'))

df.groupBy('id').count().agg(F.max("count").alias("MAX"),F.min("count").alias("MIN"),F.avg("count").alias("AVERAGE"))

5. upsert

processedDeltaTable = DeltaTable.forPath(spark, health_tracker + "processed")

update_match = """

  health_tracker.time = upserts.time 

  AND 

  health_tracker.p_device_id = upserts.p_device_id

"""

update = { "heartrate" : "upserts.heartrate" }

insert = {

  "p_device_id" : "upserts.p_device_id",

  "heartrate" : "upserts.heartrate",

  "name" : "upserts.name",

  "time" : "upserts.time",

  "dte" : "upserts.dte"

}

(processedDeltaTable.alias("health_tracker")

 .merge(upsertsDF.alias("upserts"), update_match)

 .whenMatchedUpdate(set=update)

 .whenNotMatchedInsert(values=insert)

 .execute())

%sql

MERGE INTO health_tracker_silver                            -- the MERGE instruction is used to perform the upsert

USING upserts

ON health_tracker_silver.time = upserts.time AND        

   health_tracker_silver.p_device_id = upserts.p_device_id  -- ON is used to describe the MERGE condition

WHEN MATCHED THEN                                           -- WHEN MATCHED describes the update behavior

  UPDATE SET

  health_tracker_silver.heartrate = upserts.heartrate   

WHEN NOT MATCHED THEN                                       -- WHEN NOT MATCHED describes the insert behavior

  INSERT (name, heartrate, time, dte, p_device_id)              

  VALUES (name, heartrate, time, dte, p_device_id)

6. Delta Table

display(processedDeltaTable.history())

processedDeltaTable.delete("p_device_id = 4")

Write

1. General

df.select("name", "favorite_color").write.save("namesAndFavColors.parquet")

df.write.format("json").save("s3a://...")

df.write.parquet("namesAndFavColors.parquet")

df.write.partitionBy("favorite_color").format("parquet").save("namesPartByColor.parquet")

# For file-based data source, it is also possible to bucket and sort or partition the output. 

# Bucketing and sorting are applicable only to persistent tables

df.write.partitionBy("favorite_color").bucketBy(42, "name").saveAsTable("people_partitioned_bucketed")

df.write.mode("overwrite").option("overwriteSchema", "true").format("delta").save(suppliersTargetDirectory)

2. Streaming

df.writeStream.format("delta").outputMode("append").partitionBy("p_cre8date").queryName("dax_shipment_stream")\

  .trigger(processingTime="30 seconds").option("checkpointLocation", checkpoint_path).start(destination_path)       

def processRow(row):

  # for now write them to log files, but this logic can easily be extended to publishing alerts

  # to a topic on Kafka or monitoring/paging service such as Ganglia or PagerDuty.

  print("ALERT from Sensors: Between {} and {}, device {} reported only {} times".format(row.start, row.end, row.deviceId, row[3]))

df.writeStream.outputMode("complete").foreach(processRow).start()

 def stop_all_streams() -> bool:

    stopped = False

    for stream in spark.streams.active:

        stopped = True

        stream.stop()

    return stopped

#stop all streams

streams_stopped = stop_all_streams()

if streams_stopped:

    print("All streams stopped.")

else:

    print("No running streams.")

sqm = spark.streams

[q.name for q in sqm.active]

[q.stop() for q in sqm.active]

Example

def retrieve_data(file: str) -> bool:

  """Download file from remote location to driver. Move from driver to DBFS."""

  base_url = "https://files.training.databricks.com/static/data/health-tracker/"

  url = base_url + file

  driverPath = "file:/databricks/driver/" + file

  dbfsPath   = landingPath + file

  urlretrieve(url, file)

  dbutils.fs.mv(driverPath , dbfsPath)

  return True

def load_delta_table(file: str, delta_table_path: str) -> bool:

  "Load a parquet file as a Delta table."

  parquet_df = spark.read.format("parquet").load(landingPath + file)

  parquet_df.write.format("delta").save(delta_table_path)

  return True

def process_file(file_name: str, path: str,  table_name: str) -> bool:

  """

  1. retrieve file

  2. load as delta table

  3. register table in the metastore

  """

  retrieve_data(file_name)

  print(f"Retrieve {file_name}.")

  load_delta_table(file_name, path)

  print(f"Load {file_name} to {path}")

  spark.sql(f"""

  DROP TABLE IF EXISTS {table_name}

  """)

  spark.sql(f"""

  CREATE TABLE {table_name}

  USING DELTA

  LOCATION "{path}"

  """)

  print(f"Register {table_name} using path: {path}")