""" On an abstract-level, this script illustrates several key usecases for Spark: 1. Reading files from s3 into a distributed Spark Dataframe. 2. Data transformation with User Defined Functions (UDFs). 3. Reading a Redshift query into a Spark Dataframe. 4. Performing SQL-joins between Spark Dataframes with SparkSQL. 5. Writing a Dataframe to s3. On a business-level: This script creates a mapping between Youtube content (video and asset ids), and products in our system. It does so by reading two reports available on the Youtube CMS console- Video and Asset report, into Spark Dataframes (usecase 1), and parsing the custom_claim_id column, into UPC and ISRC (usecase 2). Access to UPC and ISRC opens up a link between products in our system (usecases 3 and 4) and the granular data pulled from the Youtube Reports API (usecase 4). The relationship between the mapping tables and the data from the Reports API has the following conditions: Video report maps to Premium content within Content Owner reports where uploaderType == self Asset report maps to user-generated content (UGC) winth Asset Reports where uploaderType == thirdParty ... requiring filtering and conditional joins that can be done with ease using Spark. To run: pyspark --master local[14] --packages=com.databricks:spark-redshift_2.10:1.1.0,org.apache.hadoop:hadoop-aws:2.7.1 --jars RedshiftJDBC41-1.1.17.1017.jar,aws-java-sdk-1.11.35.jar """ import os import glob import shutil from pyspark.sql import SparkSession from pyspark.sql.functions import udf from pyspark.sql.types import StringType import utils import config import s3 s3_temp_dir = "s3n://dev-rsaporta/DataAnalyticsDept/spark/temp-dir" s3_target = "s3://dev-rsaporta/DataAnalyticsDept/spark/dumps/" data_out = "data_out/yt_test/" s3_mapping_files = 's3://dev-rsaporta/DataAnalyticsDept/youtubeAnalytics/mapping/**.csv' sc._jsc.hadoopConfiguration().set("fs.s3n.awsAccessKeyId", config.AWS_ACCESS) sc._jsc.hadoopConfiguration().set("fs.s3n.awsSecretAccessKey", config.AWS_SECRET) def clean_up(): """ Remove logs and metadata stores from past jobs. """ os.remove('derby.log') shutil.rmtree('metastore_db') junk = s3.look(s3_temp_dir.replace('s3n://','s3://')+'/*/**') if junk: for file in junk: s3.rm(file) def preprocess_mapping_file(spark, raw_mapping_file, primary_key): """ Reads a csv from s3 (raw_mapping_file), performs two user defined functions (UDFs) to transform claim_custom_id into a upc and isrc_2 column. The primary key varies between the raw_mapping files -- either video_id for Premium / self-uploaded content. -- or -- asset_id for UGC / thirdPart-uploaded content. The preprocessed file is returned as a Spark dataframe. """ df_raw_report = spark.read.format('com.databricks.spark.csv')\ .options(header='true', inferschema='true', tempdir=s3_temp_dir)\ .load(raw_mapping_file) udf_get_upc = udf(utils.custom_id_to_upc, StringType()) udf_get_isrc = udf(utils.custom_id_to_isrc, StringType()) df_preprocessed = df_raw_report.select( primary_key, 'asset_id', 'isrc', (udf_get_upc(df_raw_report['claim_custom_id'])).alias('upc'), (udf_get_isrc(df_raw_report['claim_custom_id'])).alias('isrc_2')) return df_mapping_preprocessed def join_redshift_metas(spark, df_mapping_preprocessed): """ Reads a Redshift query of release metadata (dim_release and dim_label) into a Spark Dataframe. The metadata Dataframe is then joined to the mapping dataframe. """ redshift_query = """ select a.releaseid , a.artistid , a.labelid , a.imprint , b.labelname from production.dim_release as a join ( select labelid, labelname from production.dim_label ) as b on a.labelid = b.labelid """ # query -to-> dataframe df_dim_release = spark.read \ .format("com.databricks.spark.redshift") \ .option("url", config.REDSHIFT_URL) \ .option("query", redshift_query) \ .option("tempdir", s3_temp_dir) \ .load() df_mapping_with_metadata = df_mapping_preprocessed.join( df_dim_release, df_preprocessed['upc'] == df_dim_release['releaseid'], 'leftouter').drop(df_dim_release['releaseid']) return df_mapping_with_metadata def map_yt_raw_report(spark, raw_yt_report_file, df_mapping_with_metadata, primary_key, uploader_type): """ Read a raw youtube report (gzipped CSV) into a Spark Dataframe, and joins in the metadata Dataframe. """ df_yt_raw = spark.read.format('com.databricks.spark.csv') \ .options(header='true', inferschema='true',tempdir=s3_temp_dir) \ .load(raw_yt_report_file) df_yt = df_yt_raw \ .filter(df_yt_raw['uploader_type'] == uploader_type) \ .sortWithinPartitions(primary_key) # join with the video report with product metas df_yt_final = df_yt.join( df_mapping_with_metadata, df_mapping_with_metadata[primary_key] == df_yt[primary_key], 'leftouter').drop(df_mapping_with_metadata[primary_key]) return df_yt_final def to_csv(df, local_temp_dir, s3_target): df.write.format("com.databricks.spark.csv") \ .option("path", local_temp_dir) \ .option("header","true") \ .option("delimiter","\t") \ .option("codec","gzip") \ .save() # have to use local package to dump for local_file in glob.glob(local_temp_dir+'/*.csv.gz'): s3.disk_2_s3( local_file, os.path.join(s3_target,local_file.replace(local_temp_dir,''))) os.remove(local_file) def main(): clean_up() spark = SparkSession \ .builder \ .appName(config.SPARK_APP) \ .getOrCreate() for file in s3.look(s3_mapping_files): if 'video_report' in file: primary_key = 'video_id' uploader_type = 'self' else: primary_key = 'asset_id' uploader_type = 'thirdParty' file = file.replace("s3://","s3n://") df_mapping_preprocessed = preprocess_mapping_file( spark, file, primary_key) df_mapping_with_metadata = join_redshift_metas( spark, df_mapping_preprocessed) # iterate through relevant reports goes here: to_csv(df_mapping_with_metadata, data_out, s3_target) clean_up() main()