Spark Integration - Schema for text file and column level lineage is not captured

TridevNS commented 3 months ago

Reference: https://github.com/OpenLineage/OpenLineage/issues/2573

val spark = SparkSession
  .builder()
  .appName("dataset_example").master("local[*]")
  .config("spark.extraListeners", "io.openlineage.spark.agent.OpenLineageSparkListener")
  .config("spark.openlineage.transport.type", "http")
  .config("spark.openlineage.transport.url", "http://localhost:5000")
  .config("spark.openlineage.namespace", "spark_integration")
  .config("spark.openlineage.facets.disabled", "[spark.logicalPlan]")
  .config("spark.openlineage.debugFacet", "enabled")
  .getOrCreate()

      val inputData = spark.read.option("header", "true").option("inferSchema", "true").csv("/Users/tridev/Downloads/housing.csv")

val tempDF = inputData.withColumn("total_price", col("price")*col("lotsize"))

tempDF.write.format("parquet").mode("overwrite").save("/Users/tridev/Downloads/new")

In lineage: no column lineage info like(https://openlineage.io/docs/spec/facets/dataset-facets/column_lineage_facet)

Screenshot 2024-04-15 at 12 55 45 PM Spark version: 3.3.1

In logs I can see the column level lineage:

pawel-big-lebowski commented 1 month ago

The column level lineage facet is present within the event:

[{"_producer":"https://github.com/OpenLineage/OpenLineage/tree/1.9.1/integration/spark","_schemaURL":"https://openlineage.io/spec/facets/1-0-1/ColumnLineageDatasetFacet.json#/$defs/ColumnLineageDatasetFacet","fields":{"_c0":{"inputFields":[{"namespace":"file","name":"/Users/tridev/Downloads/housing.csv","field":"_c0"}]},"price":{"inputFields":[{"namespace":"file","name":"/Users/tridev/Downloads/housing.csv","field":"price"}]},"lotsize":{"inputFields":[{"namespace":"file","name":"/Users/tridev/Downloads/housing.csv","field":"lotsize"}]},"bedrooms":{"inputFields":[{"namespace":"file","name":"/Users/tridev/Downloads/housing.csv","field":"bedrooms"}]},"bathrms":{"inputFields":[{"namespace":"file","name":"/Users/tridev/Downloads/housing.csv","field":"bathrms"}]},"stories":{"inputFields":[{"namespace":"file","name":"/Users/tridev/Downloads/housing.csv","field":"stories"}]},"driveway":{"inputFields":[{"namespace":"file","name":"/Users/tridev/Downloads/housing.csv","field":"driveway"}]},"recroom":{"inputFields":[{"namespace":"file","name":"/Users/tridev/Downloads/housing.csv","field":"recroom"}]},"fullbase":{"inputFields":[{"namespace":"file","name":"/Users/tridev/Downloads/housing.csv","field":"fullbase"}]},"gashw":{"inputFields":[{"namespace":"file","name":"/Users/tridev/Downloads/housing.csv","field":"gashw"}]},"airco":{"inputFields":[{"namespace":"file","name":"/Users/tridev/Downloads/housing.csv","field":"airco"}]},"garagepl":{"inputFields":[{"namespace":"file","name":"/Users/tridev/Downloads/housing.csv","field":"garagepl"}]},"prefarea":{"inputFields":[{"namespace":"file","name":"/Users/tridev/Downloads/housing.csv","field":"prefarea"}]},"total_price":{"inputFields":[{"namespace":"file","name":"/Users/tridev/Downloads/housing.csv","field":"price"},{"namespace":"file","name":"/Users/tridev/Downloads/housing.csv","field":"lotsize"}]}}},

and the schema is present in the events too.

Please verify if you can see in the events in Marquez lineage events tab to make sure they were successfully sent to Marquez. Then you can investigate with manually created events (like having a single column instead of 20) to find out what makes your scenario different from the one that is working properly. Of course, make sure you're using latest Marquez version.

TridevNS commented 6 days ago

producer

@pawel-big-lebowski From API how to pull the particular run event info?

We are trying to Use the metadata from a different system.

MarquezProject / marquez

Spark Integration - Schema for text file and column level lineage is not captured #2800