pyspark-etl-sample 1.0
PySpark ETL

GDSA - Deployment of Spark jobs on Azure
=========================================

# Overview
[That README](https://git.signintra.com/gdsa/induction-deployment/gdsa-induction-azure-spark/blob/master/README.md)
([on Azure DevOps](https://dev.azure.com/dbs-bigdas-sb/dbs-bigdas-d4s-cr-01/_git/gdsa-induction-azure-spark?path=%2FREADME.md&version=GBmaster)]) is part of a GitLab repository aiming to introduce how to deploy ETL-like (Extract, Transform and Load) batch jobs on Spark, managed as a service by Azure DataBricks, or deployed on Azure Kubernetes Service (AKS).

It is contributed to by the [Global Data Strategy and Analytics (GDSA) group](https://issues.atlassian.net/wiki/spaces/CDODATA/overview) and may be used well beyond by any other team.

# Table of Content (ToC)
- [GDSA - Deployment of Spark jobs on Azure](#gdsa---deployment-of-spark-jobs-on-azure)
- [Overview](#overview)
- [Table of Content (ToC)](#table-of-content--toc-)
- [Installation](#installation)
  * [Azure Databricks-CLI](#azure-databricks-cli)
- [Use cases](#use-cases)
  * [Run Spark job in local](#run-spark-job-in-local)
  * [Run Spark job in Azure Databricks](#run-spark-job-in-azure-databricks)
- [References](#references)
  * [GDSA](#gdsa)
  * [Spark](#spark)

<small><i><a href='http://ecotrust-canada.github.io/markdown-toc/'>Table of contents generated with markdown-toc</a></i></small>

# Installation
## Azure Databricks-CLI
* References:
  + [Databricks CLI](https://docs.microsoft.com/en-us/azure/databricks/dev-tools/cli/)
* Install the CLI
```bash
$ pip install databricks-cli
```
* Set up authentication
To authenticate to the CLI you use a [personal access token](https://docs.microsoft.com/en-us/azure/databricks/dev-tools/api/latest/authentication).  
To configure the CLI to use the access token:
```bash
$ databricks configure --token

Databricks Host (should begin with https://):
Token: 
```
After you complete the prompts, your access credentials are stored in the file ~/.databrickscfg.

# Use cases
## Run Spark job in local
* If not already done so, install Python 3.8.5 with PyEnv:
```bash
$ pyenv install 3.8.5 && pyenv global 3.8.5 && \
  python -mpip install -U pip && \
  python -mpip install -U wheel build tox pytest twine pyjq && \
  python -mpip install -U pandas scikit-learn && \
  python -mpip install -U pyspark && \
  pyenv global system
```
* If not already done so, clone this Git repository and go into it:
```bash
$ mkdir -p ~/dev/ml && \
  git clone git@ssh.dev.azure.com:v3/dbs-bigdas-sb/dbs-bigdas-d4s-cr-01/gdsa-induction-azure-spark ~/dev/ml/gdsa-induction-azure-spark && \
  cd ~/dev/ml/gdsa-induction-azure-spark
```
* Check Python and PySpark versions
```bash
$ pyenv version
3.8.5 (set by ~/dev/ml/gdsa-induction-azure-spark/.python-version)
$ python -V
3.8.5
$ pyspark --version
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 3.0.1
      /_/

Using Scala version 2.12.10, OpenJDK 64-Bit Server VM, 1.8.0_262
```
* Install the Python dependencies
```bash
$ pipenv install
```
* Run Spark locally with one worker thread
```bash
$ park-submit --master local --packages 'org.apache.spark:spark-avro_2.12:3.0.1' --files configs/etl_config.json --py-files packages.zip src/etl_job.py --input-data "./tests/test_data/avro_sample/50.avro"
```

## Run Spark job in Azure Databricks
[Jobs](https://docs.microsoft.com/en-us/azure/databricks/jobs) can be created in Azure Databricks workspace to run notebooks, JARs or spark code either immediately or scheduled basis. You can create and run jobs using the UI, the CLI, and by invoking the Jobs API.

* If not already done so, clone this Git repository:
```bash
$ mkdir -p ~/dev/ml && \
  git clone git@ssh.dev.azure.com:v3/dbs-bigdas-sb/dbs-bigdas-d4s-cr-01/gdsa-induction-azure-spark ~/dev/ml/gdsa-induction-azure-spark && \
  cd ~/dev/ml/gdsa-induction-azure-spark
```
* Build the PySpark codebase and dependencies 
```bash
$ make build

rm -rf dist/
mkdir ./dist
cp ./src/etl_job.py ./dist
cp -r ./tests ./dist
cp -r ./configs ./dist
cd ./src && zip -x etl_job.py -r ../dist/dependencies.zip .
  adding: dependencies/ (stored 0%)
  adding: dependencies/logging.py (deflated 63%)
  adding: dependencies/__init__.py (stored 0%)
  adding: dependencies/spark.py (deflated 61%)
```
* Create a folder for project in Databricks file system
[Databricks File System (DBFS)](https://docs.microsoft.com/en-us/azure/databricks/data/databricks-file-system) is a distributed file system mounted into an Azure Databricks workspace and available on Azure Databricks clusters.
You can upload the code and mount data and storage objects so that access data seamlessly. [DBFS CLI](https://docs.microsoft.com/en-us/azure/databricks/dev-tools/cli/dbfs-cli) is used to interact with Databricks File system in this example.
```bash
$ databricks fs mkdirs dbfs:/gdsa-induction-azure-spark
```
* Copy ./dist folder into Databricks file system 
```bash
$ dbfs cp -r ./dist dbfs:/gdsa-induction-azure-spark/dist

./dist/etl_job.py -> dbfs:/gdsa-induction-azure-spark/dist/etl_job.py
./dist/dependencies.zip -> dbfs:/gdsa-induction-azure-spark/dist/dependencies.zip
```
* Create a new Azure Databricks [job](https://docs.microsoft.com/en-us/azure/databricks/jobs) with `spark-submit` task using `databricks-job.json` file. [Jobs CLI](https://docs.microsoft.com/en-us/azure/databricks/dev-tools/cli/jobs-cli) is used to create and run jobs in this example. 
```bash
$ databricks jobs create --json-file ./databricks-job.json

{
  "job_id": 8
}
```
* To get details of the job created:
```bash
$ databricks jobs get --job-id 8

{
  "job_id": 8,
  "settings": {
    "name": "test_spark_etl",
    "new_cluster": {
      "spark_version": "7.2.x-scala2.12",
      "spark_conf": {
        "spark.databricks.delta.preview.enabled": "true"
      },
      "node_type_id": "Standard_DS3_v2",
      "spark_env_vars": {
        "PYSPARK_PYTHON": "/databricks/python3/bin/python3"
      },
      "enable_elastic_disk": true,
      "azure_attributes": {
        "availability": "ON_DEMAND_AZURE"
      },
      "num_workers": 8
    },
    "email_notifications": {},
    "timeout_seconds": 3600,
    "spark_submit_task": {
      "parameters": [
        "--py-files",
        "dbfs:/gdsa-induction-azure-spark-test/dist/dependencies.zip",
        "--files",
        "dbfs:/gdsa-induction-azure-spark-test/dist/configs/etl_config.json",
        "dbfs:/gdsa-induction-azure-spark-test/dist/etl_job.py",
	"--input-data",
	"dbfs:/gdsa-induction-azure-spark/tests/test_data/employees"
      ]
    },
    "max_concurrent_runs": 1
  },
  "created_time": 1601816306870,
  "creator_user_name": "<firstname.lastname>@dbschenker.com"
}
```
* Run the Databricks job 
```bash
$ databricks jobs run-now --job-id 8

{
  "run_id": 7,
  "number_in_job": 1
}
```
* To get the status of a running job
```bash
$ databricks runs get --run-id 7

{
  "job_id": 8,
  "run_id": 7,
  "number_in_job": 1,
  "original_attempt_run_id": 7,
  "state": {
    "life_cycle_state": "PENDING",
    "state_message": "Waiting for cluster"
  },
  "task": {
    "spark_submit_task": {
      "parameters": [
        "--py-files",
        "dbfs:/gdsa-induction-azure-spark-test/dist/dependencies.zip",
        "--files",
        "dbfs:/gdsa-induction-azure-spark-test/dist/configs/etl_config.json",
        "dbfs:/gdsa-induction-azure-spark-test/dist/etl_job.py"
      ]
    }
  },
  "cluster_spec": {
    "new_cluster": {
      "spark_version": "7.2.x-scala2.12",
      "spark_conf": {
        "spark.databricks.delta.preview.enabled": "true"
      },
      "node_type_id": "Standard_DS3_v2",
      "spark_env_vars": {
        "PYSPARK_PYTHON": "/databricks/python3/bin/python3"
      },
      "no_driver_daemon": true,
      "enable_elastic_disk": true,
      "azure_attributes": {
        "availability": "ON_DEMAND_AZURE"
      },
      "num_workers": 8
    }
  },
  "cluster_instance": {
    "cluster_id": "1004-132611-tag201"
  },
  "start_time": 1601817969522,
  "setup_duration": 0,
  "execution_duration": 0,
  "cleanup_duration": 0,
  "trigger": "ONE_TIME",
  "creator_user_name": "<firstname.lastname>@dbschenker.com",
  "run_name": "test_spark_etl",
  "run_page_url": "https://westeurope.azuredatabricks.net/?o=4656498064682730#job/8/run/1",
  "run_type": "JOB_RUN"
}
``` 
# References

## GDSA
* [GitLab - Induction to AWS deployment of EMR clusters](https://git.signintra.com/gdsa/induction-deployment/gdsa-induction-emr)
* [Confluence - Deployment of Data Pipelines](https://issues.atlassian.net/wiki/spaces/CDODATA/pages/2117239694/35+-+Deployment+of+Data+Pipelines)
  + [Confluence - Azure - Spark](https://issues.atlassian.net/wiki/spaces/CDODATA/pages/2529854143/Azure+-+Spark)
  + [Confluence - Deployment - Data Processing - Azure DataBricks Spark](https://issues.atlassian.net/wiki/spaces/CDODATA/pages/2423063978/Deployment+-+Data+Processing+-+Azure+DataBricks+Spark)
  + [Confluence - Deployment - Data Services - EMR](https://issues.atlassian.net/wiki/spaces/CDODATA/pages/1838350337/Deployment+-+Data+Services+-+EMR)
* [Confluence - Deployment of Applications](https://issues.atlassian.net/wiki/spaces/CDODATA/pages/1279787087/30+-+Deployment+of+Applications)

## Spark
* [Spark on Kubernetes](https://testdriven.io/blog/deploying-spark-on-kubernetes/),
  Jul. 2020, Michael Herman