Tracking Spark with Scala or Java

Logging metrics

You can log any custom metrics that are important for pipeline and data observability. Examples include custom metrics for data quality information, like data counts or null counts, and custom KPIs particular to your data.

To enable logging of strings and numeric values, use the ai.databand.log.DbndLogger.logMetric() method: DbndLogger.logMetric("data", data);

Tracking pipeline functions with annotations

If you have a more complex pipeline structure, or you want to present your pipeline functions and store metadata as separate tasks, you can add annotations to your pipeline code. Method annotation enables input/output tracking for each method and links them visually.

• To mark the methods that you want to track with the @Task annotation for Scala, use:

  import ai.databand.annotations.Task
  
  object ScalaSparkPipeline {
    @Task
    def main(args: Array[String]): Unit = {
      // init code
      // ...
      // task 1
  	  val imputed = unitImputation(rawData, columnsToImpute, 10)
      // task 2
    	val clean = dedupRecords(imputed, keyColumns)
      // task 3
      val report = createReport(clean)
    }
  
    @Task
    protected def unitImputation(rawData: DataFrame, columnsToImpute: Array[String], value: Int): DataFrame = {
      // ...
    }
  
    @Task
    protected def dedupRecords(data: Dataset[Row], keyColumns: Array[String]): DataFrame = {
      // ...
    }
  
    @Task
    protected def createReport(data: Dataset[Row]): Dataset[Row] = {
      // ...
    }
  }

• To mark the methods that you want to track with the @Task annotation for Java, use:

  import ai.databand.annotations.Task;
  
  public class ProcessDataSpark {
  
      @Task
      public void processCustomerData(String inputFile, String outputFile) {
          // setup code...
        	// task 1
          Dataset<Row> imputed = unitImputation(rawData, columnsToImpute, 10);
          // task 2
          Dataset<Row> clean = dedupRecords(imputed, keyColumns);
          // task 3
          Dataset<Row> report = createReport(clean);
          // ...
      }
  
      @Task
      protected Dataset<Row> unitImputation(Dataset<Row> rawData, String[] columnsToImpute, int value) {
          // ...
      }
  
      @Task
      protected Dataset<Row> dedupRecords(Dataset<Row> data, String[] keyColumns) {
          // ...
      }
  
      @Task
      protected Dataset<Row> createReport(Dataset<Row> data) {
          // ...
      }
  }
  

• To use annotations and track the flow of tasks with annotations, the Databand Java agent instruments your application and is included in the application startup script. See Installing JVM SDK and agent and Installing dbnd on Spark cluster.
• Submit your job with the following parameter:

  spark-submit ... --conf "spark.driver.extraJavaOptions=-javaagent:REPLACE_WITH_PATH_TO_AGENT/dbnd-agent-${DBND_VERSION}-all.jar

Databand can capture Spark Executor metrics, and any I/O operation by your spark code. For more information, see Tracking Spark and JVM applications.

You can enable these listeners by configuration.

To use this feature, you can add it programmatically to your Spark application. Add the Databand Spark listener to your Spark context by using Scala or Java.

• Use the following code to add the Databand Spark listener by using Scala:

  import ai.databand.annotations.Task
  import ai.databand.spark.DbndSparkListener
  import org.apache.spark.sql.{Dataset, Row, SparkSession}
  
  object CreateReport {
  
      def main(args: Array[String]): Unit = {
          val spark = SparkSession.builder
              .appName("CreateReportSparkScala")
              .getOrCreate
          val listener = new DbndSparkListener
          spark.sparkContext.addSparkListener(listener)
      }
  
  }

• Use the following code to add the Databand Spark listener by using Scala:

  import ai.databand.annotations.Task;
  import ai.databand.spark.DbndSparkListener;
  import org.apache.spark.sql.SQLContext;
  import org.apache.spark.sql.SparkSession;
  
  public class CreateReport {
  
      public static void main(String[] args) {
          SparkSession spark = SparkSession
              .builder()
              .appName("CreateReportSparkJava")
              .getOrCreate();
  
          DbndSparkListener listener = new DbndSparkListener();
          spark.sparkContext().addSparkListener(listener);
  				//...
      }
  
  }

Adding the dbnd JVM Deequ metrics repository

import ai.databand.deequ.DbndMetricsRepository

@Task
protected def dedupRecords(data: Dataset[Row], keyColumns: Array[String]): Dataset[Row] = {
    val dedupedData = data.dropDuplicates(keyColumns)
    // custom metrics repository
    val metricsRepo = new DbndMetricsRepository(new InMemoryMetricsRepository)
    // capturing dataset verification results
    VerificationSuite()
        .onData(dedupedData)
        .addCheck(
            Check(CheckLevel.Error, "Dedup testing")
                .isUnique("name")
                .isUnique("id")
                .isComplete("name")
                .isComplete("id")
                .isPositive("score"))
        .useRepository(metricsRepo)
        .saveOrAppendResult(new DbndResultKey("dedupedData"))
        .run()
    // using metrics repositoty to capture dataset profiling results
    ColumnProfilerRunner()
        .onData(dedupedData)
        .useRepository(metricsRepo)
        .saveOrAppendResult(new DbndResultKey("dedupedData"))
        .run()
}

If you already use a metrics repository, you can wrap it inside Databand's new DbndMetricsRepository(new InMemoryMetricsRepository). Databand submits the metrics to the wrapped repository first, and to the Databand tracker afterward.

To distinguish metric keys, use a special DbndResultKey. Give your checks and profiles names that you can clearly distinguish in the Databand monitoring UI.