Dedup logic in Spark SQL

Dedup logic in Spark SQL or Hive:

select
    *
from (select
    *
   ,(row_number() over (partition by user_id order by mts_trckng_rowkey)) as alias_1
    from DB_NAME.TABLE_NAME
    where dt = '20191025'
) alias_2
WHERE alias_2.alias_1 = 1;

Dedup logic in Scala:

import org.apache.spark.sql.functions._

import org.apache.spark.sql.expressions._

// Dedup logic : Remove duplicate records by eventid and DataSourceKey combination

val Data_DF_Final_Dedup = Data_DF_Final.withColumn("ROWNUM", row_number().over(Window.partitionBy(col("Id"), col("DataSourceKey")).orderBy($"Updateddate".desc))).filter("ROWNUM = 1").drop("ROWNUM")

display(Data_DF_Final_Dedup)

ADB Spark SQL

%sql 

select * from

(

select  eventid, 

row_number() OVER (PARTITION BY Id ORDER BY Updateddate DESC) alias1

from parquet.`abfss://xy@abc.dfs.core.windows.net/dataproducts/test/v1/sen/full`

where id = 115894

) alias2

where alias2.alias1=1

Spark SQL Performance Tuning

For some workloads it is possible to improve performance by either caching data in memory, or by turning on some experimental options.

Caching Data In Memory

Spark SQL can cache tables using an in-memory columnar format by calling spark.catalog.cacheTable("tableName") or dataFrame.cache(). Then Spark SQL will scan only required columns and will automatically tune compression to minimize memory usage and GC pressure. You can call spark.catalog.uncacheTable("tableName") to remove the table from memory.

Spark SQL

The entry point into all functionality in Spark is the SparkSession class. To create a basic SparkSession, just use SparkSession.builder:

from pyspark.sql import SparkSession

spark = SparkSession \
    .builder \
    .appName("Python Spark SQL basic example") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()

SparkSession in Spark 2.0 provides builtin support for Hive features including the ability to write queries using HiveQL, access to Hive UDFs, and the ability to read data from Hive tables. To use these features, you do not need to have an existing Hive setup.

With a SparkSession, applications can create DataFrames from an existing RDD, from a Hive table, or from Spark data sources.

As an example, the following creates a DataFrame based on the content of a JSON file:

# spark is an existing SparkSession
df = spark.read.json("examples/src/main/resources/people.json")
# Displays the content of the DataFrame to stdout
df.show()
# +----+-------+
# | age|   name|
# +----+-------+
# |null|Michael|
# |  30|   Andy|
# |  19| Justin|
# +----+-------+

Here we include some basic examples of structured data processing using Datasets:

In Python it’s possible to access a DataFrame’s columns either by attribute (df.age) or by indexing (df['age']). While the former is convenient for interactive data exploration, users are highly encouraged to use the latter form, which is future proof and won’t break with column names that are also attributes on the DataFrame class.

# spark, df are from the previous example
# Print the schema in a tree format
df.printSchema()
# root
# |-- age: long (nullable = true)
# |-- name: string (nullable = true)

# Select only the "name" column
df.select("name").show()
# +-------+
# |   name|
# +-------+
# |Michael|
# |   Andy|
# | Justin|
# +-------+

# Select everybody, but increment the age by 1
df.select(df['name'], df['age'] + 1).show()
# +-------+---------+
# |   name|(age + 1)|
# +-------+---------+
# |Michael|     null|
# |   Andy|       31|
# | Justin|       20|
# +-------+---------+

# Select people older than 21
df.filter(df['age'] > 21).show()
# +---+----+
# |age|name|
# +---+----+
# | 30|Andy|
# +---+----+

# Count people by age
df.groupBy("age").count().show()
# +----+-----+
# | age|count|
# +----+-----+
# |  19|    1|
# |null|    1|
# |  30|    1|
# +----+-----+

The sql function on a SparkSession enables applications to run SQL queries programmatically and returns the result as a DataFrame.

# Register the DataFrame as a SQL temporary view
df.createOrReplaceTempView("people")

sqlDF = spark.sql("SELECT * FROM people")
sqlDF.show()
# +----+-------+
# | age|   name|
# +----+-------+
# |null|Michael|
# |  30|   Andy|
# |  19| Justin|
# +----+-------+

Global Temporary View

Temporary views in Spark SQL are session-scoped and will disappear if the session that creates it terminates. If you want to have a temporary view that is shared among all sessions and keep alive until the Spark application terminates, you can create a global temporary view. Global temporary view is tied to a system preserved database global_temp, and we must use the qualified name to refer it, e.g. SELECT * FROM global_temp.view1.

# Register the DataFrame as a global temporary view
df.createGlobalTempView("people")

# Global temporary view is tied to a system preserved database `global_temp`
spark.sql("SELECT * FROM global_temp.people").show()
# +----+-------+
# | age|   name|
# +----+-------+
# |null|Michael|
# |  30|   Andy|
# |  19| Justin|
# +----+-------+

# Global temporary view is cross-session
spark.newSession().sql("SELECT * FROM global_temp.people").show()
# +----+-------+
# | age|   name|
# +----+-------+
# |null|Michael|
# |  30|   Andy|
# |  19| Justin|
# +----+-------+

Bucketing, Sorting and Partitioning

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.bucketBy(42, "name").sortBy("age").saveAsTable("people_bucketed")

while partitioning can be used with both save and saveAsTable when using the Dataset APIs.

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

It is possible to use both partitioning and bucketing for a single table:

df = spark.read.parquet("examples/src/main/resources/users.parquet")
(df
    .write
    .partitionBy("favorite_color")
    .bucketBy(42, "name")
    .saveAsTable("people_partitioned_bucketed"))

partitionBy creates a directory structure as described in the Partition Discovery section. Thus, it has limited applicability to columns with high cardinality. In contrast bucketBy distributes data across a fixed number of buckets and can be used when a number of unique values is unbounded.

Loading Data Programmatically

peopleDF = spark.read.json("examples/src/main/resources/people.json")

# DataFrames can be saved as Parquet files, maintaining the schema information.
peopleDF.write.parquet("people.parquet")

# Read in the Parquet file created above.
# Parquet files are self-describing so the schema is preserved.
# The result of loading a parquet file is also a DataFrame.
parquetFile = spark.read.parquet("people.parquet")

# Parquet files can also be used to create a temporary view and then used in SQL statements.
parquetFile.createOrReplaceTempView("parquetFile")
teenagers = spark.sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19")
teenagers.show()
# +------+
# |  name|
# +------+
# |Justin|
# +------+

Hive Tables

When working with Hive, one must instantiate SparkSession with Hive support, including connectivity to a persistent Hive metastore, support for Hive serdes, and Hive user-defined functions. Users who do not have an existing Hive deployment can still enable Hive support. When not configured by the hive-site.xml, the context automatically creates metastore_db in the current directory and creates a directory configured by spark.sql.warehouse.dir, which defaults to the directory spark-warehouse in the current directory that the Spark application is started. Note that the hive.metastore.warehouse.dir property in hive-site.xml is deprecated since Spark 2.0.0. Instead, use spark.sql.warehouse.dir to specify the default location of database in warehouse. You may need to grant write privilege to the user who starts the Spark application.

from pyspark.sql import SparkSession
from pyspark.sql import Row

# warehouse_location points to the default location for managed databases and tables
warehouse_location = abspath('spark-warehouse')

spark = SparkSession \
    .builder \
    .appName("Python Spark SQL Hive integration example") \
    .config("spark.sql.warehouse.dir", warehouse_location) \
    .enableHiveSupport() \
    .getOrCreate()

# spark is an existing SparkSession
spark.sql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING) USING hive")
spark.sql("LOAD DATA LOCAL INPATH 'examples/src/main/resources/kv1.txt' INTO TABLE src")

# Queries are expressed in HiveQL
spark.sql("SELECT * FROM src").show()
# +---+-------+
# |key|  value|
# +---+-------+
# |238|val_238|
# | 86| val_86|
# |311|val_311|
# ...

# Aggregation queries are also supported.
spark.sql("SELECT COUNT(*) FROM src").show()
# +--------+
# |count(1)|
# +--------+
# |    500 |
# +--------+

# The results of SQL queries are themselves DataFrames and support all normal functions.
sqlDF = spark.sql("SELECT key, value FROM src WHERE key < 10 ORDER BY key")

# The items in DataFrames are of type Row, which allows you to access each column by ordinal.
stringsDS = sqlDF.rdd.map(lambda row: "Key: %d, Value: %s" % (row.key, row.value))
for record in stringsDS.collect():
    print(record)
# Key: 0, Value: val_0
# Key: 0, Value: val_0
# Key: 0, Value: val_0
# ...

# You can also use DataFrames to create temporary views within a SparkSession.
Record = Row("key", "value")
recordsDF = spark.createDataFrame([Record(i, "val_" + str(i)) for i in range(1, 101)])
recordsDF.createOrReplaceTempView("records")

# Queries can then join DataFrame data with data stored in Hive.
spark.sql("SELECT * FROM records r JOIN src s ON r.key = s.key").show()
# +---+------+---+------+
# |key| value|key| value|
# +---+------+---+------+
# |  2| val_2|  2| val_2|
# |  4| val_4|  4| val_4|
# |  5| val_5|  5| val_5|
# ...

a

Spark and Hive Useful Commands

Q. How to submit pyspark script to Spark in Cluster YARN mode?

spark-submit --master yarn --deploy-mode cluster <Config_Options> <SparkScript> <PARAMETERS>

Ex:

spark-submit --master yarn --deploy-mode cluster --driver-memory 5G --conf spark.yarn.executor.extraClassPath=./ --conf spark.scheduler.mode=fair --conf spark.yarn.maxAppAttempts=1  --files /home/hadoop/hive-site.xml,s3://test-emr-bin/qa/test.config s3://test-emr-bin/qa/aggregates/emr_scripts/spark/test_aggregate_spark.py QA aggregates-qa aggregates/consume/ QA_COMM QA_AGGREGATES


Q. How to generate a hive query output to a file with columns separated by | and sorted?

hive -f src.hql > output; cat output | sed 's/\t/|/g' | sort > source;

Q. How to do you compare source and target with hive query results?

hive -f target.hql > output; cat output | sed 's/\t/|/g' | sort > target;

Q. diff source target

cat source | head -1

cat target | head -1


Q: How to update partitions with  msck repair table?

msck repair table test_daily_aggregate

Q How to show partitions

show partitions test_daily_aggregate;

Q. How to run a task in Airflow scheduler?

airflow run -i -f <DAG_NAME> <TASK_NAME> <Schedule_Date> <AIRFLOW_SCHEDULER>

airflow run -i -f demand_aggregate spark_abc_incremental_aggregate 2017-04-17 abcsla

Q. How to compile a new DAG?

airflow list_dags -sd test_dag.py 

Q. How to push a DAG?

   $ push_dag abc_data

Q. How to list Airflow dags

   $ airflow list_dags

Q. How do you pass arguments to Hive Query?

hive -f s3://abc-emr-bin/qa/pos/emr_scripts/hive/abc_transaction_lookup_stg.hive -hivevar HIVEDATABASE_STG=QA_STG_ABC -hivevar HIVEDATABASE=QA_ABC -hivevar DATE_RANGE=2017-02-10

Q. How to know size of folder in S3?

$ aws s3 ls --summarize --human-readable --recursive s3://bi-manage/dev/bi/test/global_test_daily_snapshot 

Q. How to know list of files in S3?

aws s3 ls s3://test/bi/global_test_daily_snapshot/geo_part=USA/transfer_local_snapshot_date_part=2017-05-01/ --recursive

Q. How to copy a file from local machine to EMR cluster

scp -i /Users/test/Desktop/rsa_keys/EMR.cer test.txt hadoop@10.1.1.120:/home/hadoop/

Q. How to get YARN logs

YARN logs :

---------

$ yarn logs -applicationId application_1495046596880_0037

$ yarn logs -applicationId <Application_ID>

Q. How to enable debug logging in Hive?

hive --hiveconf hive.root.logger=DEBUG,console

Q. How to know IP Address?

$ ifconfig

Q. How to add jar file to Spark (pyspark)

pyspark --jars UDFs-1.0-SNAPSHOT-jar-with-dependencies.jar;

Q. How to change execution engine in hive from Tez to mr

set hive.execution.engine=mr;

Q. How to create permanent function:

CREATE FUNCTION qa_TestFunc AS "com.org.nbac.AbcTest" using JAR 's3://emr-bin/qa/data-test/lib/UDFs-1.0-SNAPSHOT-jar-with-dependencies.jar';  

Q. How to run a java class file in a jar:

java -cp s3://emr-bin/dev/data-protection/lib/UDFs-1.0-SNAPSHOT-jar-with-dependencies.jar com.org.nbac.AbcTest.class

Q. How to source env file

$ source source.env 

$ source /Users/bac/Features/path.evn

Q. How to Build UDF Jar file using maven

$ mvn clean install -Dmaven.test.skip

Q. Basic Unit Testing of UDF in Hive:

ADD JAR  s3://-emr-bin/dev/data-test-/lib/UDFs-1.0-SNAPSHOT-jar-with-dependencies.jar;

CREATE TEMPORARY FUNCTION Encrypt AS "com.abc.test.Encrypt";

CREATE TEMPORARY FUNCTION EncryptSignature AS "com.abc.test.EncryptSignature";

CREATE TEMPORARY FUNCTION Decrypt AS "com.abc.test.Decrypt";

SELECT Encrypt('abc', 'xyz', "dev_test");

Q. Basic Unit Testing of UDF in Spark:

spark.sql('CREATE TEMPORARY FUNCTION Encrypt AS "com.abc.test.Encrypt"')

spark.sql('CREATE TEMPORARY FUNCTION EncryptSignature AS "com.abc.test.EncryptSignature"')

spark.sql('CREATE TEMPORARY FUNCTION Decrypt AS "com.abc.test.Decrypt"')

spark.sql('select source_id,EncryptSignature(source_id, current_status, "dev_activity_status") as sourceid_signature from stg_inactivity')

spark-submit --master yarn --deploy-mode cluster --executor-cores 5 --num-executors 90 --driver-memory 5GB --executor-memory 10G --conf spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version=2 --conf fs.s3n.multipart.uploads.enabled=true --conf spark.dynamicAllocation.enabled=false --conf spark.yarn.executor.memoryOverhead=5120 --jars s3://test-emr-bin/dev/test-protection/lib/Test_UDFs-1.0-SNAPSHOT-jar-with-dependencies.jar test_encrypt.py

Q. How to add partitions in Hive

ALTER TABLE dev_test ADD  PARTITION 's3n://bi-managed/dev/test/process_date=2017-05-31'

Q. Insert Overwrite table in Hive

INSERT OVERWRITE TABLE dev_list

SELECT member_id, email_addr

FROM dev_test

;

Q. How to create an external table in Hive

DROP TABLE IF EXISTS dev_list;

CREATE EXTERNAL TABLE `dev_list`(

  `source_id` string, 

  `sourceid_signature` string, 

  `source_cd` string, 

  `status` string)

STORED AS PARQUET

LOCATION

  's3://bi-managed/prod/test/data/2017-06-12'

;

Q. How to submit a job in Spark 

spark-submit --jars s3://emr-bin/dev/common/emr_scripts/jars/UDFs-1.0-SNAPSHOT-jar-with-dependencies.jar pysark_job.py bi-managed/prod bi-managed/prod 2017-06-12

Q. How to set JAVA compiler in Maven to JDK 1.7 

-------------

<properties>

    <maven.compiler.source>1.7</maven.compiler.source>

    <maven.compiler.target>1.7</maven.compiler.target>

  </properties>

Ref: https://maven.apache.org/plugins/maven-compiler-plugin/examples/set-compiler-source-and-target.html

--------------------------

Q. How to get avro schema from given avro data file

java -jar ~/Downloads/avro-tools-1.7.4.jar getschema part-r-00000-c7fc5e23-842e-4723-86e1-c069afdeb7c1.avro > chn_pos_clean_stage.avsc

Q. How is use of coalesce() function in Spark SQL

public static Column coalesce(Column... e)

Returns the first column that is not null, or null if all inputs are null.

For example, coalesce(a, b, c) will return a if a is not null, or b if a is null and b is not null, or c if both a and b are null but c is not null.

Q. What is use of concat_ws() function in Spark SQL

public static Column concat_ws(java.lang.String sep, Column... exprs)

Concatenates multiple input string columns together into a single string column, using the given separator.