# delphi-spark-poc ## Streams Aggregation (`src/streams_agg.py`) This job reads decompressed json data from AWS S3, performs stream aggregations per date and write data back to AWS S3 in parquet format. #### Local execution ##### Using `spark-submit` ```bash $ spark-submit --master local[*] src/streams_agg.py ``` Example: ```bash $ spark-submit --master local[*] src/streams_agg.py \ --dsp spotify \ -d 1970-01-01 \ -s s3a://path/to/source \ -t s3a://path/to/target ``` ##### Using Make The following command will create new python virtual environment if it is not exists, install all required dependencies and starts the spark job. `ARGS` variable is used for providing all application properties. ```bash $ make ARGS="..." run-streams-agg ``` Example: ```bash $ make ARGS="--dsp spotify -d 1970-01-01 -s s3a://path/to/source -t s3a://path/to/target" run-streams-agg ``` The possible arguments are: | key | required? | description | default value | |--------------------------|-----------|---------------------------------------------------------------------------------------------------|---------------| | `--dsp` | true | A list of DSP providers. Supports multiple values separated by space. | | | `-d`, `--date` | true | A list of dates in a following format: `YYYY-MM-DD`. Supports multiple values separated by space. | | | `-t`, `--target-folder` | true | A target directory to write parquet files. | | | `-s`, `--source-folder` | true | A source directory root to json files. | | | `-l`, `--licensor` | false | A list of licensors to process. Supports multiple values separated by space. | * | | `-c`, `--country` | false | A list of countries to process. Supports multiple values separated by space. | * | ##### Providing S3 credentials: ###### Using `spark-submit`. It's possible to provide AWS credentials using `fs.s3a.access.key` and `fs.s3a.secret.key` properties as shown in example below. ```bash $ spark-submit \ --conf fs.s3a.access.key= \ --conf fs.s3a.secret.key= \ --master local[*] \ streams_agg.py ``` ###### Using env vars. There is an alternative way to provide AWS credentials by configuring environment variables. ```bash $ export AWS_ACCESS_KEY_ID= $ export AWS_SECRET_ACCESS_KEY= ``` #### Starting job on EMR Cluster configuration (64 workers x i3.4xlarge): ``` --driver-cores 5 --driver-memory 31G --executor-cores 5 --executor-memory 31G --num-executors 191 --conf spark.default.parallelism=3820 --conf spark.sql.shuffle.partitions=3820 ``` AWS CLI run command: ```bash $ aws emr add-steps \ --cluster-id \ --steps Type=spark,Name=DelphiSparkPOC,Args=[--deploy-mode,cluster,--master,yarn,--driver-cores,5,--driver-memory,31G,--num-executors,191,--executor-cores,5,--executor-memory,31G,--conf,spark.yarn.submit.waitAppCompletion=false,--conf,fs.s3a.access.key=,--conf,fs.s3a.secret.key=,--packages,org.apache.hadoop:hadoop-aws:2.7.3,--conf,spark.sql.shuffle.partitions=3820,--conf,spark.default.parallelism=3820,--conf,spark.executor.extraJavaOptions=-XX:+UseG1GC,s3://delphi-spark-poc/streams_agg.py,--dsp,spotify,-d,2019-05-29],ActionOnFailure=CONTINUE ``` #### Starting job on Databricks: Cluster configuration (64 workers x i3.4xlarge): ``` --conf spark.default.parallelism=3820 --conf spark.sql.shuffle.partitions=3820 --conf spark.executor.extraJavaOptions=-XX:+UseG1GC ``` REST API call: ```bash $ curl -n \ -X POST \ -H 'Content-Type: application/json' \ -H "Authorization: Bearer " \ -d streams_agg.py \ https://dbc-b2991935-97ea.cloud.databricks.com/api/2.0/jobs/run-now ``` #### Starting job on QuBole: Cluster configuration (64 workers x i3.4xlarge): ``` --driver-cores 5 --driver-memory 31G --executor-cores 5 --executor-memory 31G --num-executors 191 --conf spark.default.parallelism=3820 --conf spark.sql.shuffle.partitions=3820 ``` REST API call: ```bash $ curl -i \ -X POST \ -H "X-AUTH-TOKEN: " \ -H "Content-Type: application/json" \ -H "Accept: application/json" \ -d '{"script_location":"s3://dev-delphi-qubole/streams_agg.py", "arguments":"--driver-cores 5 --driver-memory 31G --num-executors 191 --executor-cores 5 --executor-memory 31G --conf spark.sql.shuffle.partitions=3820 --conf spark.default.parallelism=3820", "user_program_arguments" : "--dsp spotify -d 2019-05-29", "language":"python", "command_type":"SparkCommand", "label":""}' \ https://us.qubole.com/api/v1.2/commands ``` ## Streams Aggregation Validation (`src/streams_agg_validation.py`) This job reads outputs from `src/streams_agg.py` and compares results for schema equality and data equality. #### Local execution ##### Using `spark-submit` ```bash $ spark-submit --master local[*] src/streams_agg_validation.py ``` Example: ```bash $ spark-submit --master local[*] src/streams_agg_validation.py --urls s3a://path/to/df1 s3a://path/to/df2 ``` ##### Using Make The following command will create new python virtual environment if it is not exists, install all required dependencies and starts the spark job. `ARGS` variable is used for providing all application properties. ```bash $ make ARGS="..." run-streams-agg-validation ``` Example: ```bash $ make ARGS="--urls s3a://path/to/df1 s3a://path/to/df2" run-streams-agg-validation ``` Supported arguments are: | key | required? | description | |----------|-----------|------------------------------------------------------------------------------------------------------------------------------------------------| | `--urls` | true | A list of urls to output data frames. At least two data frames must be provided. `*` is supported at the end of url: `s3a://path/to/folder/*`. |