End-to-End-Beispiel für die Verwendung von Databand-APIs

Das folgende Beispiel zeigt die Verwendung von dbnd APIs.

example.py*

import time
import logging
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from dbnd import dbnd_tracking, task, dataset_op_logger, log_metric, log_dataframe

# Initialize Spark session
spark = SparkSession.builder \
    .appName("Data Pipeline with Databand") \
    .getOrCreate()

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@task
def create_sample_data():
    # Create a DataFrame with sample data including columns to be dropped
    data = [
        ("John", "Camping Equipment", 500, "Regular", "USA"),
        ("Jane", "Golf Equipment", 300, "Premium", "UK"),
        ("Mike", "Camping Equipment", 450, "Regular", "USA"),
        ("Emily", "Golf Equipment", 350, "Premium", "Canada"),
        ("Anna", "Camping Equipment", 600, "Regular", "USA"),
        ("Tom", "Golf Equipment", 200, "Regular", "UK")
    ]
    columns = ["Name", "Product line", "Sales", "Customer Type", "Country"]
    retailData = spark.createDataFrame(data, columns)

    # Log the data creation
    unique_file_name = "sample-data"
    with dataset_op_logger(unique_file_name, "read", with_schema=True, with_preview=True, with_stats=True) as logger:
        logger.set(data=retailData)

    return retailData

@task
def filter_data(rawData):
    # Define columns to drop
    columns_to_drop = ['Customer Type', 'Country']

    # Drop the specified columns in PySpark DataFrame
    filteredRetailData = rawData.drop(*columns_to_drop)

    # Log the data after dropping columns
    unique_file_name = 'script://Weekly_Sales/Filtered_df'
    with dataset_op_logger(unique_file_name, "read", with_schema=True, with_preview=True) as logger:
        logger.set(data=filteredRetailData)

    return filteredRetailData

@task
def write_data_by_product_line(filteredData):
    # Filter data for Camping Equipment and write to CSV
    campingEquipment = filteredData.filter(col('Product line') == 'Camping Equipment')
    campingEquipment.write.csv("Camping_Equipment.csv", header=True, mode="overwrite")

    # Log writing the Camping Equipment CSV
    log_dataframe("camping_equipment", campingEquipment, with_schema=True, with_stats=True)

    # Filter data for Golf Equipment and write to CSV
    golfEquipment = filteredData.filter(col('Product line') == 'Golf Equipment')
    golfEquipment.write.csv("Golf_Equipment.csv", header=True, mode="overwrite")

    # Log writing the Golf Equipment CSV
    log_dataframe("golf_equipment", golfEquipment, with_schema=True, with_stats=True)

def prepare_retail_data():
    with dbnd_tracking(
            conf={
                "tracking": {
                    "track_source_code": True
                },
                "log": {
                    "preview_head_bytes": 15360,
                    "preview_tail_bytes": 15360
                }
            }
    ):

        logger.info("Running Databand spark application!")

        start_time_milliseconds = int(round(time.time() * 1000))
        log_metric("metric_check", "OK")

        # Call the step job - create sample data
        rawData = create_sample_data()

        # Filter data
        filteredData = filter_data(rawData)

        # Write data by product line
        write_data_by_product_line(filteredData)

        end_time_milliseconds = int(round(time.time() * 1000))
        elapsed_time = end_time_milliseconds - start_time_milliseconds

        log_metric('elapsed-time', elapsed_time)

        logger.info(f"Total pipeline running time: {elapsed_time:.2f} milliseconds")
        logger.info("Spark execution completed..")

        log_metric("is-success", "OK")

# Invoke the main function
prepare_retail_data()

Kurzer Überblick über die verwendeten Databand APIs und Funktionen:

• dbnd_tracking: Initialisiert die Verfolgung für Ihre Pipeline oder Anwendung, konfiguriert die Databand-Einstellungen und protokolliert Ausführungsdetails. Verwendung:

  with dbnd_tracking(conf={...}, job_name="job_name", run_name="run_name"):
   # Pipeline code
  

• task: Markiert eine Funktion als Databand-Aufgabe und ermöglicht so die Verfolgung und Überwachung einzelner Schritte in Ihrer Pipeline.

  Verwendung:

  @task
  def my_task_function():
  

• dataset_op_logger: Protokolliert Operationen auf Datensätzen, einschließlich Schema und Statistiken.

  Verwendung:

     with dataset_op_logger(dataset_name, operation_type) as logger:
         logger.set(data=my_dataframe)
  

• log_metric: Zeichnet benutzerdefinierte Metriken auf, um die Leistung oder andere quantitative Daten während der Ausführung zu verfolgen.

  Verwendung:

  log_metric("metric_name", metric_value)
  

• log_dataframe: Protokolliert Details über eine DataFrame, wie Schema und Statistiken zur Überwachung von Datentransformationen.

  Verwendung:

  log_dataframe("dataframe_name", my_dataframe, with_schema=True, with_stats=True)
  

Informationen zum Einreichen von Spark-Aufträgen finden Sie unter Einreichen von Engine-Anwendungen.

Nachdem Sie den Spark-Antrag eingereicht haben, erhalten Sie eine Bestätigungsnachricht mit der Antrags-ID und der Spark-Version. Bewahren Sie diese Informationen auf, um den Ausführungsstatus Ihres eingereichten Spark-Auftrags zu verfolgen. Sie können Datensätze mit Hilfe der Tracking-Funktionen von Databand innerhalb der Databand-Umgebung überwachen und verfolgen.

Um das Databand-Dashboard für die Nachverfolgung anzuzeigen, gehen Sie zu Konfigurationen > IBM Data Observability by Databand und klicken Sie auf View Databand.

Weitere Informationen und Beispiele finden Sie unter:

• IBM Beobachtbarkeit der Daten nach Datenband
• Für PySpark Anwendungen: Verfolgung PySpark
• Für Spark( Java / Scala ): Verfolgung von Spark(Scala / Java)