Installing on a Spark cluster

Setting Databand configuration by using environment variables

You need to define the environment variables at the API call or EmrCreateJobFlowOperator Airflow Operator. Alternatively, you can provide these variables from AWS UI when you create a new cluster. The EMR cluster doesn't have a way of defining environment variables in the bootstrap. Consult with the official EMR documentation on Spark Configuration if you use a custom operator or if you create clusters outside of Airflow.

from airflow.hooks.base_hook import BaseHook

dbnd_config = BaseHook.get_connection("dbnd_config").extra_dejson
databand_url = dbnd_config["core"]["databand_url"]
databand_access_token = dbnd_config["core"]["databand_access_token"]

emr_create_job_flow = EmrCreateJobFlowOperator(
    job_flow_overrides={
        "Name": "<EMR Cluster Name>",
        #...
        "Configurations": [
            {
                "Classification": "spark-env",
                "Configurations": [
                    {
                        "Classification": "export",
                        "Properties": {
                            "DBND__TRACKING": "True",
                            "DBND__ENABLE__SPARK_CONTEXT_ENV": "True",
                            "DBND__CORE__DATABAND_URL": databand_url,
                            "DBND__CORE__DATABAND_ACCESS_TOKEN": databand_access_token,
                        },
                    }
                ],
            }
        ]
    }
    #...
)

Installing Databand on cluster

As the EMR cluster has support for bootstrap actions, the following snippet can be used to install Python and JVM integrations:

#!/usr/bin/env bash

DBND_VERSION=REPLACE_WITH_DBND_VERSION

sudo python3 -m pip install pandas==1.2.0 pydeequ==1.0.1 databand==${DBND_VERSION}
sudo wget https://repo1.maven.org/maven2/ai/databand/dbnd-agent/${DBND_VERSION}/dbnd-agent-${DBND_VERSION}-all.jar -P /home/hadoop/

Add this script to your cluster bootstrap actions list. For more information, see bootstrap actions documentation.

Setting Databand configuration with environment variables

In the cluster configuration screen, click Edit > Advanced Options > Spark. Inside the Environment Variables section, declare the listed configuration variables. Be sure to replace <databand-url> and <databand-access-token> with your environment-specific information:

• DBND__TRACKING="True"
• DBND__ENABLE__SPARK_CONTEXT_ENV="True"
• DBND__CORE__DATABAND_URL="REPLACE_WITH_DATABAND_URL"
• DBND__CORE__DATABAND_ACCESS_TOKEN="REPLACE_WITH_DATABAND_TOKEN"

Install Python dbnd library in Databricks cluster

Under the Libraries tab of your cluster's configuration:

1. Click Install New.
2. Choose PyPI.
3. Enter databand==REPLACE_WITH_DBND_VERSION as the Package name.
4. Click Install.

Install Python dbnd library for specific Airflow operator

Do not use this mode in production, use it only for trying out dbnd in specific tasks. Make sure that the dbnd library is installed on the Databricks cluster by adding databand to the libraries parameter of the DatabricksSubmitRunOperator, as shown in the following example:

DatabricksSubmitRunOperator(
     #...
     json={"libraries": [
        {"pypi": {"package": "databand==REPLACE_WITH_DBND_VERSION"}},
    ]},
)

Tracking Scala or Java Spark jobs

Download dbnd agent and place it into your DBFS working folder.

To configure Tracking Spark applications with automatic dataset logging, add ai.databand.spark.DbndSparkQueryExecutionListener as a spark.sql.queryExecutionListeners and add ai.databand.spark.DbndSparkListener as a spark.extraListeners (this mode works only if you enable the dbnd agent).

Use the following configuration of the Databricks job to enable Databand Java Agent with automatic dataset tracking:

spark_operator = DatabricksSubmitRunOperator(
    json={
        #...
        "new_cluster": {
            #...
               "spark_conf": {
                "spark.sql.queryExecutionListeners": "ai.databand.spark.DbndSparkQueryExecutionListener",
                "spark.extraListeners": "ai.databand.spark.DbndSparkListener",
                "spark.driver.extraJavaOptions": "-javaagent:/dbfs/apps/dbnd-agent-0.xx.x-all.jar",
            },
        },
        #...
    })

First, make sure that you publish the agent to /dbfs/apps/. For more configuration options, see the Databricks Runs Submit API documentation.

Cluster setup

You can define environment variables during the cluster setup or add these variables to your bootstrap as described in the previous sections about Spark clusters.

from airflow.hooks.base_hook import BaseHook

dbnd_config = BaseHook.get_connection("dbnd_config").extra_dejson
databand_url = dbnd_config["core"]["databand_url"]
databand_access_token = dbnd_config["core"]["databand_access_token"]

cluster_create = DataprocClusterCreateOperator(
     # ...
     properties={
        "spark-env:DBND__TRACKING": "True",
        "spark-env:DBND__ENABLE__SPARK_CONTEXT_ENV": "True",
        "spark-env:DBND__CORE__DATABAND_URL": databand_url,
        "spark-env:DBND__CORE__DATABAND_ACCESS_TOKEN": databand_access_token,
    },
    # ...
)

You can install Databand PySpark support by using the same operator:

cluster_create = DataprocClusterCreateOperator(
     #...
     properties={
             "dataproc:pip.packages": "dbnd==REPLACE_WITH_DATABAND_VERSION", 
     }
     #...
)

The Dataproc cluster supports initialization. The following script installs Databand libraries and sets up the environment variables that are required for tracking:

#!/usr/bin/env bash

DBND_VERSION=REPLACE_WITH_DBND_VERSION

## to use conda-provided python instead of system one
export PATH=/opt/conda/default/bin:${PATH}

python3 -m pip install pydeequ==1.0.1 databand==${DBND_VERSION}

DBND__CORE__DATABAND_ACCESS_TOKEN=$(/usr/share/google/get_metadata_value attributes/DBND__CORE__DATABAND_ACCESS_TOKEN)
sh -c "echo DBND__CORE__DATABAND_ACCESS_TOKEN=${DBND__CORE__DATABAND_ACCESS_TOKEN} >> /usr/lib/spark/conf/spark-env.sh"
DBND__CORE__DATABAND_URL=$(/usr/share/google/get_metadata_value attributes/DBND__CORE__DATABAND_URL)
sh -c "echo DBND__CORE__DATABAND_URL=${DBND__CORE__DATABAND_URL} >> /usr/lib/spark/conf/spark-env.sh"
sh -c "echo DBND__TRACKING=True >> /usr/lib/spark/conf/spark-env.sh"
sh -c "echo DBND__ENABLE__SPARK_CONTEXT_ENV=True >> /usr/lib/spark/conf/spark-env.sh"

Note: Variables like access token and tracker url can be passed to the initialization action through cluster metadata properties. For more information, see official Dataproc documentation.

Tracking Python Spark jobs

Use the following configuration of the PySpark DataProc job to enable Databand Spark query listener with automatic dataset tracking:

pyspark_operator = DataProcPySparkOperator(
    #...
    dataproc_pyspark_jars=[ "gs://.../dbnd-agent-REPLACE_WITH_DATABAND_VERSION-all.jar"],
    dataproc_pyspark_properties={
        "spark.sql.queryExecutionListeners": "ai.databand.spark.DbndSparkQueryExecutionListener",
        "spark.extraListeners": "ai.databand.spark.DbndSparkListener",
    },
    #...
)

• First, publish your JAR to Google Storage.

For the list of all supported operators and extra information, see Tracking remote tasks.