DataJob Entity

The DataJob entity represents a unit of work in a data processing pipeline (e.g., an Airflow task, a dbt model, a Spark job). DataJobs belong to DataFlows (pipelines) and can have lineage to datasets and other DataJobs. This guide covers comprehensive DataJob operations in SDK V2.

Creating a DataJob

Minimal DataJob

Orchestrator, flowId, and jobId are required:

DataJob dataJob = DataJob.builder()
    .orchestrator("airflow")
    .flowId("my_dag")
    .jobId("my_task")
    .build();

With Cluster

Specify cluster (default is "prod"):

DataJob dataJob = DataJob.builder()
    .orchestrator("airflow")
    .flowId("analytics_pipeline")
    .cluster("staging")
    .jobId("transform_data")
    .build();
// URN: urn:li:dataJob:(urn:li:dataFlow:(airflow,analytics_pipeline,staging),transform_data)

With Metadata

Add description and name at construction (requires both name AND type):

DataJob dataJob = DataJob.builder()
    .orchestrator("dagster")
    .flowId("customer_etl")
    .cluster("prod")
    .jobId("load_customers")
    .description("Loads customer data from PostgreSQL to Snowflake")
    .name("Load Customers to DWH")
    .type("BATCH")
    .build();

With Custom Properties

Include custom properties in builder (requires name and type when using customProperties):

Map<String, String> props = new HashMap<>();
props.put("schedule", "0 2 * * *");
props.put("retries", "3");
props.put("timeout", "3600");

DataJob dataJob = DataJob.builder()
    .orchestrator("airflow")
    .flowId("daily_pipeline")
    .jobId("my_task")
    .name("My Daily Task")
    .type("BATCH")
    .customProperties(props)
    .build();

URN Construction

DataJob URNs follow the pattern:

urn:li:dataJob:(urn:li:dataFlow:({orchestrator},{flowId},{cluster}),{jobId})

Automatic URN creation:

DataJob dataJob = DataJob.builder()
    .orchestrator("airflow")
    .flowId("finance_reporting")
    .cluster("prod")
    .jobId("aggregate_transactions")
    .build();

DataJobUrn urn = dataJob.getDataJobUrn();
// urn:li:dataJob:(urn:li:dataFlow:(airflow,finance_reporting,prod),aggregate_transactions)

Description Operations

Setting Description

dataJob.setDescription("Processes daily customer transactions");

Reading Description

Get description (lazy-loaded from DataJobInfo):

String description = dataJob.getDescription();

Display Name Operations

Setting Name

dataJob.setName("Process Customer Transactions");

Reading Name

String name = dataJob.getName();

Tags

Adding Tags

// Simple tag name (auto-prefixed)
dataJob.addTag("critical");
// Creates: urn:li:tag:critical

// Full tag URN
dataJob.addTag("urn:li:tag:etl");

Removing Tags

dataJob.removeTag("critical");
dataJob.removeTag("urn:li:tag:etl");

Tag Chaining

dataJob.addTag("critical")
       .addTag("pii")
       .addTag("production");

Owners

Adding Owners

import com.linkedin.common.OwnershipType;

// Technical owner
dataJob.addOwner(
    "urn:li:corpuser:data_team",
    OwnershipType.TECHNICAL_OWNER
);

// Data steward
dataJob.addOwner(
    "urn:li:corpuser:compliance",
    OwnershipType.DATA_STEWARD
);

// Business owner
dataJob.addOwner(
    "urn:li:corpuser:product_team",
    OwnershipType.BUSINESS_OWNER
);

Removing Owners

dataJob.removeOwner("urn:li:corpuser:data_team");

Owner Types

Available ownership types:

• TECHNICAL_OWNER - Maintains the technical implementation
• BUSINESS_OWNER - Business stakeholder
• DATA_STEWARD - Manages data quality and compliance
• DATAOWNER - Generic data owner
• DEVELOPER - Software developer
• PRODUCER - Data producer
• CONSUMER - Data consumer
• STAKEHOLDER - Other stakeholder

Glossary Terms

Adding Terms

dataJob.addTerm("urn:li:glossaryTerm:DataProcessing");
dataJob.addTerm("urn:li:glossaryTerm:ETL");

Removing Terms

dataJob.removeTerm("urn:li:glossaryTerm:DataProcessing");

Term Chaining

dataJob.addTerm("urn:li:glossaryTerm:DataProcessing")
       .addTerm("urn:li:glossaryTerm:ETL")
       .addTerm("urn:li:glossaryTerm:FinancialReporting");

Domain

Setting Domain

dataJob.setDomain("urn:li:domain:Engineering");

Removing Domain

dataJob.removeDomain();

Custom Properties

Adding Individual Properties

dataJob.addCustomProperty("schedule", "0 2 * * *");
dataJob.addCustomProperty("retries", "3");
dataJob.addCustomProperty("timeout", "3600");

Setting All Properties

Replace all custom properties:

Map<String, String> properties = new HashMap<>();
properties.put("schedule", "0 2 * * *");
properties.put("retries", "3");
properties.put("timeout", "3600");
properties.put("priority", "high");

dataJob.setCustomProperties(properties);

Removing Properties

dataJob.removeCustomProperty("timeout");

Lineage Operations

DataJob lineage defines the relationship between data jobs and the datasets they operate on. Lineage enables impact analysis, data provenance tracking, and understanding data flows through your pipelines.

The DataJob SDK supports four types of lineage:

1. Dataset-level lineage - Track which datasets a job reads from and writes to
2. DataJob dependencies - Track which jobs depend on other jobs (task dependencies)
3. Field-level lineage - Track specific columns consumed and produced
4. Fine-grained lineage - Track column-to-column transformations

Understanding Input and Output Datasets

Input Datasets - Datasets that the job reads from:

• Represent source data for the job
• Create upstream lineage: Dataset → DataJob

Output Datasets - Datasets that the job writes to:

• Represent destination data from the job
• Create downstream lineage: DataJob → Dataset

Input Datasets

Adding Single Inlet

// Using string URN
dataJob.addInputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.transactions,PROD)");

// Using DatasetUrn object for type safety
DatasetUrn datasetUrn = DatasetUrn.createFromString(
    "urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.transactions,PROD)"
);
dataJob.addInputDataset(datasetUrn);

Adding Multiple Inlets

// Chain multiple calls
dataJob.addInputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.transactions,PROD)")
       .addInputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.customers,PROD)")
       .addInputDataset("urn:li:dataset:(urn:li:dataPlatform:kafka,events.purchases,PROD)");

Setting All Inlets at Once

List<String> inletUrns = Arrays.asList(
    "urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.orders,PROD)",
    "urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.customers,PROD)",
    "urn:li:dataset:(urn:li:dataPlatform:kafka,events.clicks,PROD)"
);
dataJob.setInputDatasets(inletUrns);

Removing Inlets

// Remove single inlet
dataJob.removeInputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.transactions,PROD)");

// Or using DatasetUrn
DatasetUrn datasetUrn = DatasetUrn.createFromString(
    "urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.transactions,PROD)"
);
dataJob.removeInputDataset(datasetUrn);

Reading Inlets

// Get all inlets (lazy-loaded)
List<DatasetUrn> inlets = dataJob.getInputDatasets();
for (DatasetUrn inlet : inlets) {
    System.out.println("Input: " + inlet);
}

Output Datasets (Outlets)

Adding Single Outlet

// Using string URN
dataJob.addOutputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.sales_summary,PROD)");

// Using DatasetUrn object
DatasetUrn datasetUrn = DatasetUrn.createFromString(
    "urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.sales_summary,PROD)"
);
dataJob.addOutputDataset(datasetUrn);

Adding Multiple Outlets

dataJob.addOutputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.daily_summary,PROD)")
       .addOutputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.monthly_summary,PROD)")
       .addOutputDataset("urn:li:dataset:(urn:li:dataPlatform:s3,reports/summary.parquet,PROD)");

Setting All Outlets at Once

List<String> outletUrns = Arrays.asList(
    "urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.customer_metrics,PROD)",
    "urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.product_metrics,PROD)"
);
dataJob.setOutputDatasets(outletUrns);

Removing Outlets

// Remove single outlet
dataJob.removeOutputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.sales_summary,PROD)");

// Or using DatasetUrn
DatasetUrn datasetUrn = DatasetUrn.createFromString(
    "urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.sales_summary,PROD)"
);
dataJob.removeOutputDataset(datasetUrn);

Reading Outlets

// Get all outlets (lazy-loaded)
List<DatasetUrn> outlets = dataJob.getOutputDatasets();
for (DatasetUrn outlet : outlets) {
    System.out.println("Output: " + outlet);
}

DataJob Dependencies

DataJob dependencies model task-to-task relationships within workflows. This enables DataHub to track which jobs depend on other jobs completing first.

Use cases:

• Airflow task dependencies (task A → task B → task C)
• Cross-DAG dependencies (jobs in different pipelines)
• Workflow orchestration visualization

Adding Job Dependencies

// Using string URN
dataJob.addInputDataJob("urn:li:dataJob:(urn:li:dataFlow:(airflow,pipeline,prod),upstream_task)");

// Using DataJobUrn object for type safety
DataJobUrn upstreamJob = DataJobUrn.createFromString(
    "urn:li:dataJob:(urn:li:dataFlow:(airflow,pipeline,prod),upstream_task)"
);
dataJob.addInputDataJob(upstreamJob);

Chaining Job Dependencies

// Multiple dependencies (task runs after all complete)
dataJob.addInputDataJob("urn:li:dataJob:(urn:li:dataFlow:(airflow,pipeline,prod),task_1)")
       .addInputDataJob("urn:li:dataJob:(urn:li:dataFlow:(airflow,pipeline,prod),task_2)")
       .addInputDataJob("urn:li:dataJob:(urn:li:dataFlow:(dagster,other_pipeline,prod),external_task)");

Removing Job Dependencies

// Remove single dependency
dataJob.removeInputDataJob("urn:li:dataJob:(urn:li:dataFlow:(airflow,pipeline,prod),task_1)");

// Or using DataJobUrn
DataJobUrn jobUrn = DataJobUrn.createFromString(
    "urn:li:dataJob:(urn:li:dataFlow:(airflow,pipeline,prod),task_1)"
);
dataJob.removeInputDataJob(jobUrn);

Reading Job Dependencies

// Get all upstream job dependencies (lazy-loaded)
List<DataJobUrn> dependencies = dataJob.getInputDataJobs();
for (DataJobUrn dependency : dependencies) {
    System.out.println("Depends on: " + dependency);
}

Example: Airflow Task Dependencies

// Model a typical Airflow DAG task chain
DataJob extractTask = DataJob.builder()
    .orchestrator("airflow")
    .flowId("etl_pipeline")
    .jobId("extract_data")
    .build();

DataJob validateTask = DataJob.builder()
    .orchestrator("airflow")
    .flowId("etl_pipeline")
    .jobId("validate_data")
    .build();

// validate_data depends on extract_data
validateTask.addInputDataJob(extractTask.getUrn().toString());

DataJob transformTask = DataJob.builder()
    .orchestrator("airflow")
    .flowId("etl_pipeline")
    .jobId("transform_data")
    .build();

// transform_data depends on validate_data
transformTask.addInputDataJob(validateTask.getUrn().toString());

// Save all tasks
client.entities().upsert(extractTask);
client.entities().upsert(validateTask);
client.entities().upsert(transformTask);

// Result: extract_data → validate_data → transform_data

Field-Level Lineage

Field-level lineage tracks which specific columns (fields) a job consumes and produces. This provides finer granularity than dataset-level lineage.

Use cases:

• Track which columns are read/written by transformations
• Understand field-level dependencies
• Validate that jobs only access necessary columns

Field URN Format:

urn:li:schemaField:(DATASET_URN,COLUMN_NAME)

Adding Input Fields

// Track which columns the job reads
dataJob.addInputField("urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,db.orders,PROD),order_id)");
dataJob.addInputField("urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,db.orders,PROD),customer_id)");
dataJob.addInputField("urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,db.orders,PROD),total_amount)");

Adding Output Fields

// Track which columns the job writes
dataJob.addOutputField("urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.sales,PROD),order_id)");
dataJob.addOutputField("urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.sales,PROD),customer_id)");
dataJob.addOutputField("urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.sales,PROD),revenue)");

Removing Fields

// Remove field lineage
dataJob.removeInputField("urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,db.orders,PROD),order_id)");
dataJob.removeOutputField("urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.sales,PROD),revenue)");

Reading Fields

// Get all input fields (lazy-loaded)
List<Urn> inputFields = dataJob.getInputFields();
for (Urn field : inputFields) {
    System.out.println("Reads field: " + field);
}

// Get all output fields (lazy-loaded)
List<Urn> outputFields = dataJob.getOutputFields();
for (Urn field : outputFields) {
    System.out.println("Writes field: " + field);
}

Example: Column-Level Tracking

DataJob aggregateJob = DataJob.builder()
    .orchestrator("airflow")
    .flowId("analytics")
    .jobId("aggregate_sales")
    .description("Aggregates sales data by customer")
    .name("Aggregate Sales by Customer")
    .type("BATCH")
    .build();

// Dataset-level lineage
aggregateJob.addInputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.transactions,PROD)");
aggregateJob.addOutputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.customer_sales,PROD)");

// Field-level lineage - specify exact columns used
aggregateJob.addInputField("urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.transactions,PROD),customer_id)");
aggregateJob.addInputField("urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.transactions,PROD),amount)");
aggregateJob.addInputField("urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.transactions,PROD),transaction_date)");

aggregateJob.addOutputField("urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.customer_sales,PROD),customer_id)");
aggregateJob.addOutputField("urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.customer_sales,PROD),total_sales)");
aggregateJob.addOutputField("urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.customer_sales,PROD),transaction_count)");

client.entities().upsert(aggregateJob);

Fine-Grained Lineage

Fine-grained lineage captures column-to-column transformations, showing exactly which input columns produce which output columns and how they're transformed.

Use cases:

• Document transformation logic (e.g., "SUM(amount)")
• Track column-level impact analysis
• Understand data derivations
• Compliance and audit trails

Adding Fine-Grained Lineage

// Basic transformation (no confidence score)
dataJob.addFineGrainedLineage(
    "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.orders,PROD),customer_id)",
    "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.sales,PROD),customer_id)",
    "IDENTITY",
    null
);

// Transformation with confidence score (0.0 to 1.0)
dataJob.addFineGrainedLineage(
    "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.orders,PROD),amount)",
    "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.sales,PROD),revenue)",
    "SUM",
    1.0f  // High confidence
);

Common Transformation Types

// IDENTITY - direct copy
dataJob.addFineGrainedLineage(upstream, downstream, "IDENTITY", 1.0f);

// Aggregations
dataJob.addFineGrainedLineage(upstream, downstream, "SUM", 1.0f);
dataJob.addFineGrainedLineage(upstream, downstream, "COUNT", 1.0f);
dataJob.addFineGrainedLineage(upstream, downstream, "AVG", 1.0f);
dataJob.addFineGrainedLineage(upstream, downstream, "MAX", 1.0f);
dataJob.addFineGrainedLineage(upstream, downstream, "MIN", 1.0f);

// String operations
dataJob.addFineGrainedLineage(upstream, downstream, "CONCAT", 0.9f);
dataJob.addFineGrainedLineage(upstream, downstream, "UPPER", 1.0f);
dataJob.addFineGrainedLineage(upstream, downstream, "SUBSTRING", 0.95f);

// Date operations
dataJob.addFineGrainedLineage(upstream, downstream, "DATE_TRUNC", 1.0f);
dataJob.addFineGrainedLineage(upstream, downstream, "EXTRACT", 1.0f);

// Custom transformations
dataJob.addFineGrainedLineage(upstream, downstream, "CUSTOM_FUNCTION", 0.8f);

Removing Fine-Grained Lineage

// Remove specific transformation
dataJob.removeFineGrainedLineage(
    "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.orders,PROD),amount)",
    "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.sales,PROD),revenue)",
    "SUM",
    null  // queryUrn (optional)
);

Reading Fine-Grained Lineage

// Get all fine-grained lineage (lazy-loaded)
List<FineGrainedLineage> lineages = dataJob.getFineGrainedLineages();
for (FineGrainedLineage lineage : lineages) {
    System.out.println("Upstreams: " + lineage.getUpstreams());
    System.out.println("Downstreams: " + lineage.getDownstreams());
    System.out.println("Transformation: " + lineage.getTransformOperation());
    System.out.println("Confidence: " + lineage.getConfidenceScore());
}

Example: Complex Aggregation

DataJob salesAggregation = DataJob.builder()
    .orchestrator("airflow")
    .flowId("analytics")
    .jobId("daily_sales_summary")
    .name("Daily Sales Summary")
    .type("BATCH")
    .build();

// Dataset-level lineage
salesAggregation.addInputDataset("urn:li:dataset:(urn:li:dataPlatform:postgres,sales.transactions,PROD)");
salesAggregation.addOutputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.daily_summary,PROD)");

// Fine-grained transformations
String inputDataset = "urn:li:dataset:(urn:li:dataPlatform:postgres,sales.transactions,PROD)";
String outputDataset = "urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.daily_summary,PROD)";

// Date is copied directly
salesAggregation.addFineGrainedLineage(
    "urn:li:schemaField:(" + inputDataset + ",transaction_date)",
    "urn:li:schemaField:(" + outputDataset + ",date)",
    "IDENTITY",
    1.0f
);

// Revenue is SUM of amounts
salesAggregation.addFineGrainedLineage(
    "urn:li:schemaField:(" + inputDataset + ",amount)",
    "urn:li:schemaField:(" + outputDataset + ",total_revenue)",
    "SUM",
    1.0f
);

// Transaction count
salesAggregation.addFineGrainedLineage(
    "urn:li:schemaField:(" + inputDataset + ",transaction_id)",
    "urn:li:schemaField:(" + outputDataset + ",transaction_count)",
    "COUNT",
    1.0f
);

// Average order value
salesAggregation.addFineGrainedLineage(
    "urn:li:schemaField:(" + inputDataset + ",amount)",
    "urn:li:schemaField:(" + outputDataset + ",avg_order_value)",
    "AVG",
    1.0f
);

client.entities().upsert(salesAggregation);

Example: Multi-Column Derivation

// Model a transformation where output depends on multiple input columns
DataJob enrichmentJob = DataJob.builder()
    .orchestrator("airflow")
    .flowId("enrichment")
    .jobId("enrich_customer_data")
    .build();

String inputDataset = "urn:li:dataset:(urn:li:dataPlatform:postgres,crm.customers,PROD)";
String outputDataset = "urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.customers_enriched,PROD)";

// full_name = CONCAT(first_name, ' ', last_name)
// Both first_name and last_name contribute to full_name
enrichmentJob.addFineGrainedLineage(
    "urn:li:schemaField:(" + inputDataset + ",first_name)",
    "urn:li:schemaField:(" + outputDataset + ",full_name)",
    "CONCAT",
    1.0f
);

enrichmentJob.addFineGrainedLineage(
    "urn:li:schemaField:(" + inputDataset + ",last_name)",
    "urn:li:schemaField:(" + outputDataset + ",full_name)",
    "CONCAT",
    1.0f
);

// email_domain = SUBSTRING(email, POSITION('@', email) + 1)
enrichmentJob.addFineGrainedLineage(
    "urn:li:schemaField:(" + inputDataset + ",email)",
    "urn:li:schemaField:(" + outputDataset + ",email_domain)",
    "SUBSTRING",
    1.0f
);

client.entities().upsert(enrichmentJob);

Confidence Scores

Confidence scores (0.0 to 1.0) indicate how certain you are about the transformation:

• 1.0 - Exact, deterministic transformation (e.g., IDENTITY, SUM)
• 0.9-0.99 - High confidence (e.g., simple string operations)
• 0.7-0.89 - Medium confidence (e.g., complex transformations with some uncertainty)
• 0.5-0.69 - Low confidence (e.g., ML-derived lineage, heuristic-based)
• < 0.5 - Very uncertain (generally not recommended)

// High confidence - exact transformation known
dataJob.addFineGrainedLineage(source, target, "UPPER", 1.0f);

// Medium confidence - inferred from SQL parsing
dataJob.addFineGrainedLineage(source, target, "CASE_WHEN", 0.85f);

// Low confidence - ML-predicted transformation
dataJob.addFineGrainedLineage(source, target, "INFERRED", 0.6f);

Complete Lineage Example

This example demonstrates all four types of lineage working together:

// Create upstream validation job
DataJob validateJob = DataJob.builder()
    .orchestrator("airflow")
    .flowId("analytics_pipeline")
    .cluster("prod")
    .jobId("validate_transactions")
    .name("Validate Transaction Data")
    .type("BATCH")
    .build();

validateJob.addInputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.transactions,PROD)")
           .addOutputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,validated.transactions,PROD)");

client.entities().upsert(validateJob);

// Create main transformation job with comprehensive lineage
DataJob transformJob = DataJob.builder()
    .orchestrator("airflow")
    .flowId("analytics_pipeline")
    .cluster("prod")
    .jobId("aggregate_sales")
    .description("Aggregates daily sales data from multiple validated sources")
    .name("Aggregate Daily Sales")
    .type("BATCH")
    .build();

// 1. Dataset-level lineage - Which tables are read/written
transformJob.addInputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,validated.transactions,PROD)")
            .addInputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.customers,PROD)")
            .addOutputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.daily_sales,PROD)");

// 2. DataJob dependencies - This job depends on the validation job
transformJob.addInputDataJob(validateJob.getUrn().toString());

// 3. Field-level lineage - Which specific columns are accessed
String transactionsDataset = "urn:li:dataset:(urn:li:dataPlatform:snowflake,validated.transactions,PROD)";
String customersDataset = "urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.customers,PROD)";
String outputDataset = "urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.daily_sales,PROD)";

// Input fields
transformJob.addInputField("urn:li:schemaField:(" + transactionsDataset + ",transaction_id)")
            .addInputField("urn:li:schemaField:(" + transactionsDataset + ",customer_id)")
            .addInputField("urn:li:schemaField:(" + transactionsDataset + ",amount)")
            .addInputField("urn:li:schemaField:(" + transactionsDataset + ",transaction_date)")
            .addInputField("urn:li:schemaField:(" + customersDataset + ",customer_id)")
            .addInputField("urn:li:schemaField:(" + customersDataset + ",customer_name)");

// Output fields
transformJob.addOutputField("urn:li:schemaField:(" + outputDataset + ",date)")
            .addOutputField("urn:li:schemaField:(" + outputDataset + ",customer_name)")
            .addOutputField("urn:li:schemaField:(" + outputDataset + ",total_revenue)")
            .addOutputField("urn:li:schemaField:(" + outputDataset + ",transaction_count)");

// 4. Fine-grained lineage - Specific column-to-column transformations
// Date column (identity transformation)
transformJob.addFineGrainedLineage(
    "urn:li:schemaField:(" + transactionsDataset + ",transaction_date)",
    "urn:li:schemaField:(" + outputDataset + ",date)",
    "IDENTITY",
    1.0f
);

// Customer name (join + identity)
transformJob.addFineGrainedLineage(
    "urn:li:schemaField:(" + customersDataset + ",customer_name)",
    "urn:li:schemaField:(" + outputDataset + ",customer_name)",
    "IDENTITY",
    1.0f
);

// Total revenue (aggregation)
transformJob.addFineGrainedLineage(
    "urn:li:schemaField:(" + transactionsDataset + ",amount)",
    "urn:li:schemaField:(" + outputDataset + ",total_revenue)",
    "SUM",
    1.0f
);

// Transaction count (aggregation)
transformJob.addFineGrainedLineage(
    "urn:li:schemaField:(" + transactionsDataset + ",transaction_id)",
    "urn:li:schemaField:(" + outputDataset + ",transaction_count)",
    "COUNT",
    1.0f
);

// Add other metadata
transformJob.addTag("critical")
            .addOwner("urn:li:corpuser:data_team", OwnershipType.TECHNICAL_OWNER);

// Save to DataHub
client.entities().upsert(transformJob);

// Result: Creates comprehensive lineage showing:
// - Job dependency: validate_transactions → aggregate_sales
// - Dataset flow: raw.transactions → validated.transactions → analytics.daily_sales
//                 raw.customers → analytics.daily_sales
// - Column-level: transaction_date → date (IDENTITY)
//                 amount → total_revenue (SUM)
//                 transaction_id → transaction_count (COUNT)
//                 customer_name → customer_name (IDENTITY via JOIN)

Lineage Flow Visualization

The comprehensive lineage example above creates this multi-level lineage graph:

Job-to-Job Level:
┌────────────────────────┐         ┌──────────────────────┐
│ Validate Transactions  │────────→│  Aggregate Sales Job │
└────────────────────────┘         └──────────────────────┘

Dataset Level:
┌─────────────────────┐    ┌─────────────────────────┐    ┌─────────────────────────┐
│ raw.transactions    │───→│ validated.transactions  │───→│                         │
└─────────────────────┘    └─────────────────────────┘    │  analytics.daily_sales  │
                                                           │                         │
┌─────────────────────┐                                   │                         │
│ raw.customers       │──────────────────────────────────→│                         │
└─────────────────────┘                                   └─────────────────────────┘

Column Level (Fine-Grained):
validated.transactions.transaction_date ──[IDENTITY]──→ daily_sales.date
validated.transactions.amount           ──[SUM]──────→ daily_sales.total_revenue
validated.transactions.transaction_id   ──[COUNT]────→ daily_sales.transaction_count
raw.customers.customer_name             ──[IDENTITY]──→ daily_sales.customer_name

ETL Pipeline Example

Model a complete Extract-Transform-Load pipeline:

// Extract job
DataJob extractJob = DataJob.builder()
    .orchestrator("airflow")
    .flowId("etl_pipeline")
    .jobId("extract")
    .build();

extractJob.addInputDataset("urn:li:dataset:(urn:li:dataPlatform:mysql,prod.orders,PROD)")
          .addOutputDataset("urn:li:dataset:(urn:li:dataPlatform:s3,staging/orders_raw,PROD)");

client.entities().upsert(extractJob);

// Transform job
DataJob transformJob = DataJob.builder()
    .orchestrator("airflow")
    .flowId("etl_pipeline")
    .jobId("transform")
    .build();

transformJob.addInputDataset("urn:li:dataset:(urn:li:dataPlatform:s3,staging/orders_raw,PROD)")
            .addOutputDataset("urn:li:dataset:(urn:li:dataPlatform:s3,staging/orders_clean,PROD)");

client.entities().upsert(transformJob);

// Load job
DataJob loadJob = DataJob.builder()
    .orchestrator("airflow")
    .flowId("etl_pipeline")
    .jobId("load")
    .build();

loadJob.addInputDataset("urn:li:dataset:(urn:li:dataPlatform:s3,staging/orders_clean,PROD)")
       .addOutputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.orders,PROD)");

client.entities().upsert(loadJob);

// Creates end-to-end lineage:
// mysql.orders → [Extract] → s3.raw → [Transform] → s3.clean → [Load] → snowflake.analytics

Updating Lineage

// Load existing job
DataJobUrn urn = DataJobUrn.createFromString(
    "urn:li:dataJob:(urn:li:dataFlow:(airflow,my_pipeline,prod),my_task)"
);
DataJob dataJob = client.entities().get(urn);

// Add new inlet (e.g., requirements changed)
dataJob.addInputDataset("urn:li:dataset:(urn:li:dataPlatform:kafka,events.new_source,PROD)");

// Remove old outlet (e.g., deprecated table)
dataJob.removeOutputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,old.deprecated_table,PROD)");

// Apply changes
client.entities().update(dataJob);

Lineage Best Practices

1. Be Complete - Define both inputs and outputs for accurate lineage
2. Use Correct URNs - Ensure dataset URNs match existing datasets in DataHub
3. Update When Changed - Keep lineage current as pipelines evolve
4. Document Transformations - Use descriptions to explain what the job does
5. Model All Jobs - Include every step in your pipeline for complete lineage
6. Use Typed URNs - Prefer DatasetUrn/DataJobUrn objects over strings for compile-time safety
7. Layer Your Lineage - Start with dataset-level, add field-level and fine-grained as needed
8. Track Dependencies - Use DataJob dependencies to model task orchestration
9. Be Precise with Transformations - Use accurate transformation types in fine-grained lineage
10. Set Confidence Scores - Use appropriate confidence scores to indicate lineage quality

Common Patterns

Multiple Sources to Single Destination

// Data aggregation job
dataJob.addInputDataset("urn:li:dataset:(urn:li:dataPlatform:postgres,sales.orders,PROD)")
       .addInputDataset("urn:li:dataset:(urn:li:dataPlatform:postgres,sales.customers,PROD)")
       .addInputDataset("urn:li:dataset:(urn:li:dataPlatform:postgres,sales.products,PROD)")
       .addOutputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.sales_summary,PROD)");

Single Source to Multiple Destinations

// Data fanout job
dataJob.addInputDataset("urn:li:dataset:(urn:li:dataPlatform:kafka,events.raw,PROD)")
       .addOutputDataset("urn:li:dataset:(urn:li:dataPlatform:s3,archive/events,PROD)")
       .addOutputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,events.processed,PROD)")
       .addOutputDataset("urn:li:dataset:(urn:li:dataPlatform:elasticsearch,events.searchable,PROD)");

Cross-Platform Lineage

// ETL across different platforms
dataJob.addInputDataset("urn:li:dataset:(urn:li:dataPlatform:mysql,production.transactions,PROD)")
       .addInputDataset("urn:li:dataset:(urn:li:dataPlatform:kafka,events.user_activity,PROD)")
       .addInputDataset("urn:li:dataset:(urn:li:dataPlatform:s3,raw/reference_data,PROD)")
       .addOutputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.customer_360,PROD)")
       .addOutputDataset("urn:li:dataset:(urn:li:dataPlatform:bigquery,reporting.customer_metrics,PROD)");

Complete Example

import datahub.client.v2.DataHubClientV2;
import datahub.client.v2.entity.DataJob;
import com.linkedin.common.OwnershipType;
import java.io.IOException;
import java.util.concurrent.ExecutionException;

public class DataJobExample {
    public static void main(String[] args) {
        // Create client
        DataHubClientV2 client = DataHubClientV2.builder()
            .server("http://localhost:8080")
            .build();

        try {
            // Build data job with all metadata
            DataJob dataJob = DataJob.builder()
                .orchestrator("airflow")
                .flowId("customer_analytics")
                .cluster("prod")
                .jobId("process_events")
                .description("Processes customer events from Kafka to warehouse")
                .name("Process Customer Events")
                .type("BATCH")
                .build();

            // Add tags
            dataJob.addTag("critical")
                   .addTag("etl")
                   .addTag("pii");

            // Add owners
            dataJob.addOwner("urn:li:corpuser:data_team", OwnershipType.TECHNICAL_OWNER)
                   .addOwner("urn:li:corpuser:product_team", OwnershipType.BUSINESS_OWNER);

            // Add glossary terms
            dataJob.addTerm("urn:li:glossaryTerm:DataProcessing")
                   .addTerm("urn:li:glossaryTerm:CustomerData");

            // Set domain
            dataJob.setDomain("urn:li:domain:Analytics");

            // Add custom properties
            dataJob.addCustomProperty("schedule", "0 2 * * *")
                   .addCustomProperty("retries", "3")
                   .addCustomProperty("timeout", "7200");

            // Upsert to DataHub
            client.entities().upsert(dataJob);

            System.out.println("Successfully created data job: " + dataJob.getUrn());

        } catch (IOException | ExecutionException | InterruptedException e) {
            e.printStackTrace();
        } finally {
            try {
                client.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

Updating Existing DataJobs

Load and Modify

// Load existing data job
DataJobUrn urn = DataJobUrn.createFromString(
    "urn:li:dataJob:(urn:li:dataFlow:(airflow,my_dag,prod),my_task)"
);
DataJob dataJob = client.entities().get(urn);

// Add new metadata (creates patches)
dataJob.addTag("new-tag")
       .addOwner("urn:li:corpuser:new_owner", OwnershipType.TECHNICAL_OWNER);

// Apply patches
client.entities().update(dataJob);

Incremental Updates

// Just add what you need
dataJob.addTag("critical");
client.entities().update(dataJob);

// Later, add more
dataJob.addCustomProperty("priority", "high");
client.entities().update(dataJob);

Builder Options Reference

Method	Required	Description
orchestrator(String)	✅ Yes	Orchestrator (e.g., "airflow", "dagster")
flowId(String)	✅ Yes	Flow/DAG identifier
jobId(String)	✅ Yes	Job/task identifier
cluster(String)	No	Cluster name (e.g., "prod", "dev"). Default: "prod"
description(String)	No	Job description. Requires both name() and type() to be set
name(String)	No	Display name shown in UI. Required if using description(), type(), or customProperties()
type(String)	No	Job type (e.g., "BATCH", "STREAMING"). Required if using description(), name(), or customProperties()
customProperties(Map)	No	Map of custom key-value properties. Requires both name() and type() to be set

Important: The DataJobInfo aspect requires both name and type fields. If you provide any of description, name, type, or customProperties in the builder, you must provide both name and type. Otherwise, you'll get an IllegalArgumentException at build time.

Common Patterns

Creating Multiple DataJobs

String[] tasks = {"extract", "transform", "load"};
for (String taskName : tasks) {
    DataJob dataJob = DataJob.builder()
        .orchestrator("airflow")
        .flowId("etl_pipeline")
        .cluster("prod")
        .jobId(taskName)
        .build();

    dataJob.addTag("etl")
           .addCustomProperty("team", "data-engineering");

    client.entities().upsert(dataJob);
}

Batch Metadata Addition

DataJob dataJob = DataJob.builder()
    .orchestrator("airflow")
    .flowId("my_dag")
    .jobId("my_task")
    .build();

List<String> tags = Arrays.asList("critical", "production", "etl");
tags.forEach(dataJob::addTag);

client.entities().upsert(dataJob);  // Emits all tags in one call

Conditional Metadata

if (isCritical(dataJob)) {
    dataJob.addTag("critical")
           .addTerm("urn:li:glossaryTerm:BusinessCritical");
}

if (processesFinancialData(dataJob)) {
    dataJob.addTag("financial")
           .addOwner("urn:li:corpuser:compliance_team", OwnershipType.DATA_STEWARD);
}

DataJob vs DataFlow

DataFlow represents a pipeline or DAG (e.g., an Airflow DAG):

• URN: urn:li:dataFlow:(orchestrator,flowId,cluster)
• Contains multiple DataJobs

DataJob represents a task within a pipeline:

• URN: urn:li:dataJob:(flowUrn,jobId)
• Belongs to one DataFlow
• Can have lineage to datasets and other DataJobs

Example hierarchy:

DataFlow: urn:li:dataFlow:(airflow,customer_pipeline,prod)
├── DataJob: urn:li:dataJob:(urn:li:dataFlow:(airflow,customer_pipeline,prod),extract)
├── DataJob: urn:li:dataJob:(urn:li:dataFlow:(airflow,customer_pipeline,prod),transform)
└── DataJob: urn:li:dataJob:(urn:li:dataFlow:(airflow,customer_pipeline,prod),load)

Orchestrator Examples

Common orchestrator values:

• airflow - Apache Airflow
• dagster - Dagster
• prefect - Prefect
• dbt - dbt (data build tool)
• spark - Apache Spark
• glue - AWS Glue
• dataflow - Google Cloud Dataflow
• azkaban - Azkaban
• luigi - Luigi

Next Steps

• Dataset Entity - Working with dataset entities
• Patch Operations - Deep dive into patches
• Migration Guide - Upgrading from V1

Examples

Basic DataJob Creation

# Inlined from /metadata-integration/java/examples/src/main/java/io/datahubproject/examples/v2/DataJobCreateExample.java
package io.datahubproject.examples.v2;

import com.linkedin.common.OwnershipType;
import datahub.client.v2.DataHubClientV2;
import datahub.client.v2.entity.DataJob;
import java.io.IOException;
import java.util.concurrent.ExecutionException;

/**
 * Example demonstrating how to create a DataJob using Java SDK V2.
 *
 * <p>This example shows:
 *
 * <ul>
 *   <li>Creating a DataHubClientV2
 *   <li>Building a DataJob with fluent builder
 *   <li>Adding tags, owners, and custom properties
 *   <li>Upserting to DataHub
 * </ul>
 */
public class DataJobCreateExample {

  public static void main(String[] args)
      throws IOException, ExecutionException, InterruptedException {
    // Create client (use environment variables or pass explicit values)
    DataHubClientV2 client =
        DataHubClientV2.builder()
            .server(System.getenv().getOrDefault("DATAHUB_SERVER", "http://localhost:8080"))
            .token(System.getenv("DATAHUB_TOKEN")) // Optional
            .build();

    try {
      // Test connection
      if (!client.testConnection()) {
        System.err.println("Failed to connect to DataHub server");
        return;
      }
      System.out.println("✓ Connected to DataHub");

      // Build data job with metadata
      DataJob dataJob =
          DataJob.builder()
              .orchestrator("airflow")
              .flowId("user_analytics_dag")
              .cluster("prod")
              .jobId("process_user_events")
              .description("Processes user interaction events and loads into data warehouse")
              .name("Process User Events")
              .type("BATCH")
              .build();

      System.out.println("✓ Built data job with URN: " + dataJob.getUrn());

      // Add tags
      dataJob.addTag("critical").addTag("etl").addTag("user-data");

      System.out.println("✓ Added 3 tags");

      // Add owners
      dataJob
          .addOwner("urn:li:corpuser:datahub", OwnershipType.TECHNICAL_OWNER)
          .addOwner("urn:li:corpuser:data_team", OwnershipType.DATA_STEWARD);

      System.out.println("✓ Added 2 owners");

      // Add custom properties
      dataJob
          .addCustomProperty("team", "data-engineering")
          .addCustomProperty("schedule", "0 2 * * *")
          .addCustomProperty("retries", "3")
          .addCustomProperty("timeout", "3600");

      System.out.println("✓ Added 4 custom properties");

      // Upsert to DataHub
      client.entities().upsert(dataJob);

      System.out.println("✓ Successfully created data job in DataHub!");
      System.out.println("\n  URN: " + dataJob.getUrn());
      System.out.println(
          "  View in DataHub: " + client.getConfig().getServer() + "/dataJob/" + dataJob.getUrn());

    } finally {
      client.close();
    }
  }
}

Comprehensive DataJob with Metadata and Lineage

# Inlined from /metadata-integration/java/examples/src/main/java/io/datahubproject/examples/v2/DataJobFullExample.java
package io.datahubproject.examples.v2;

import com.linkedin.common.OwnershipType;
import datahub.client.v2.DataHubClientV2;
import datahub.client.v2.entity.DataJob;
import java.io.IOException;
import java.util.concurrent.ExecutionException;

/**
 * Comprehensive example demonstrating all DataJob metadata operations using Java SDK V2.
 *
 * <p>This example shows:
 *
 * <ul>
 *   <li>Creating a data job with complete metadata
 *   <li>Adding tags, owners, glossary terms
 *   <li>Setting domain and custom properties
 *   <li>Defining lineage with input datasets (inlets) and output datasets (outlets)
 *   <li>Combining all operations in single entity
 * </ul>
 */
public class DataJobFullExample {

  public static void main(String[] args)
      throws IOException, ExecutionException, InterruptedException {
    // Create client
    DataHubClientV2 client =
        DataHubClientV2.builder()
            .server(System.getenv().getOrDefault("DATAHUB_SERVER", "http://localhost:8080"))
            .token(System.getenv("DATAHUB_TOKEN"))
            .build();

    try {
      // Test connection
      if (!client.testConnection()) {
        System.err.println("Failed to connect to DataHub server");
        return;
      }
      System.out.println("✓ Connected to DataHub");

      // Build comprehensive data job with all metadata types
      DataJob dataJob =
          DataJob.builder()
              .orchestrator("airflow")
              .flowId("financial_reporting_pipeline")
              .cluster("prod")
              .jobId("aggregate_customer_transactions")
              .description(
                  "Critical ETL job that aggregates daily customer transaction data from multiple sources "
                      + "and loads into the enterprise data warehouse. Includes data quality checks, "
                      + "PII tokenization, and regulatory compliance validation.")
              .name("Aggregate Customer Transactions")
              .type("BATCH")
              .build();

      System.out.println("✓ Built data job with URN: " + dataJob.getUrn());

      // Add multiple tags for categorization
      dataJob
          .addTag("critical")
          .addTag("pii")
          .addTag("financial")
          .addTag("etl")
          .addTag("production");

      System.out.println("✓ Added 5 tags");

      // Add multiple owners with different roles
      dataJob
          .addOwner("urn:li:corpuser:data_engineering", OwnershipType.TECHNICAL_OWNER)
          .addOwner("urn:li:corpuser:finance_team", OwnershipType.BUSINESS_OWNER)
          .addOwner("urn:li:corpuser:compliance_team", OwnershipType.DATA_STEWARD);

      System.out.println("✓ Added 3 owners");

      // Add glossary terms for business context
      dataJob
          .addTerm("urn:li:glossaryTerm:DataProcessing")
          .addTerm("urn:li:glossaryTerm:ETL")
          .addTerm("urn:li:glossaryTerm:FinancialReporting");

      System.out.println("✓ Added 3 glossary terms");

      // Set domain for organizational structure
      dataJob.setDomain("urn:li:domain:Finance");

      System.out.println("✓ Set domain");

      // Add comprehensive custom properties
      dataJob
          .addCustomProperty("team", "data-platform")
          .addCustomProperty("schedule", "0 2 * * *") // Daily at 2 AM
          .addCustomProperty("retries", "3")
          .addCustomProperty("timeout", "7200") // 2 hours
          .addCustomProperty("sla_hours", "4")
          .addCustomProperty("priority", "high")
          .addCustomProperty("notification_channel", "#data-alerts")
          .addCustomProperty("requires_manual_approval", "false")
          .addCustomProperty("data_classification", "highly-confidential")
          .addCustomProperty("compliance_level", "PCI-DSS");

      System.out.println("✓ Added 10 custom properties");

      // Add lineage: Define input datasets (inlets) that this job reads from
      dataJob
          .addInputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.transactions,PROD)")
          .addInputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.customers,PROD)")
          .addInputDataset("urn:li:dataset:(urn:li:dataPlatform:kafka,events.user_activity,PROD)");

      System.out.println("✓ Added 3 input datasets (inlets)");

      // Add lineage: Define output datasets (outlets) that this job writes to
      dataJob
          .addOutputDataset(
              "urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.customer_transactions,PROD)")
          .addOutputDataset(
              "urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.daily_summary,PROD)");

      System.out.println("✓ Added 2 output datasets (outlets)");

      // Count accumulated patches
      System.out.println("\nAccumulated " + dataJob.getPendingPatches().size() + " patches");

      // Upsert to DataHub - all metadata in single operation
      client.entities().upsert(dataJob);

      System.out.println("\n✓ Successfully created comprehensive data job in DataHub!");
      System.out.println("\nSummary:");
      System.out.println("  URN: " + dataJob.getUrn());
      System.out.println("  Orchestrator: airflow");
      System.out.println("  Flow: financial_reporting_pipeline");
      System.out.println("  Job: aggregate_customer_transactions");
      System.out.println("  Tags: 5");
      System.out.println("  Owners: 3");
      System.out.println("  Glossary Terms: 3");
      System.out.println("  Domain: Finance");
      System.out.println("  Custom Properties: 10");
      System.out.println("  Input Datasets (Inlets): 3");
      System.out.println("  Output Datasets (Outlets): 2");
      System.out.println(
          "\n  View in DataHub: "
              + client.getConfig().getServer()
              + "/dataJob/"
              + dataJob.getUrn());
      System.out.println("\nLineage Flow:");
      System.out.println(
          "  raw.transactions, raw.customers, events.user_activity → [Job] → customer_transactions, daily_summary");

    } finally {
      client.close();
    }
  }
}

DataJob Lineage Operations

# Inlined from /metadata-integration/java/examples/src/main/java/io/datahubproject/examples/v2/DataJobLineageExample.java
package io.datahubproject.examples.v2;

import com.linkedin.common.urn.DatasetUrn;
import datahub.client.v2.DataHubClientV2;
import datahub.client.v2.entity.DataJob;
import java.io.IOException;
import java.net.URISyntaxException;
import java.util.Arrays;
import java.util.concurrent.ExecutionException;

/**
 * Focused example demonstrating DataJob lineage operations using Java SDK V2.
 *
 * <p>This example shows:
 *
 * <ul>
 *   <li>Creating a DataJob with input datasets and output datasets
 *   <li>Adding and removing individual inputs and outputs
 *   <li>Setting multiple inputs/outputs at once
 *   <li>Building a complete ETL pipeline with lineage
 *   <li>Understanding the relationship between DataJob and DataFlow
 * </ul>
 *
 * <p>Lineage Concepts:
 *
 * <ul>
 *   <li><b>Input Datasets</b> - Datasets that the job reads from
 *   <li><b>Output Datasets</b> - Datasets that the job writes to
 *   <li>Lineage flows: Dataset (input) → DataJob → Dataset (output)
 * </ul>
 */
public class DataJobLineageExample {

  public static void main(String[] args)
      throws IOException, ExecutionException, InterruptedException, URISyntaxException {
    // Create client
    DataHubClientV2 client =
        DataHubClientV2.builder()
            .server(System.getenv().getOrDefault("DATAHUB_SERVER", "http://localhost:8080"))
            .token(System.getenv("DATAHUB_TOKEN"))
            .build();

    try {
      // Test connection
      if (!client.testConnection()) {
        System.err.println("Failed to connect to DataHub server");
        return;
      }
      System.out.println("✓ Connected to DataHub\n");

      // ==================== Example 1: Simple ETL Job with Lineage ====================
      System.out.println("Example 1: Creating ETL job with lineage");
      System.out.println("==========================================");

      DataJob etlJob =
          DataJob.builder()
              .orchestrator("airflow")
              .flowId("customer_etl_pipeline")
              .cluster("prod")
              .jobId("load_customers")
              .description("Extracts customer data from PostgreSQL and loads into Snowflake")
              .name("Load Customers")
              .type("BATCH")
              .build();

      // Add input dataset (source)
      etlJob.addInputDataset("urn:li:dataset:(urn:li:dataPlatform:postgres,public.customers,PROD)");

      // Add output dataset (destination)
      etlJob.addOutputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,dwh.customers,PROD)");

      client.entities().upsert(etlJob);

      System.out.println("✓ Created ETL job: " + etlJob.getUrn());
      System.out.println(
          "  Lineage: postgres.public.customers → [Load Customers] → snowflake.dwh.customers\n");

      // ==================== Example 2: Complex Aggregation Job ====================
      System.out.println("Example 2: Complex aggregation with multiple inputs/outputs");
      System.out.println("============================================================");

      DataJob aggregationJob =
          DataJob.builder()
              .orchestrator("airflow")
              .flowId("analytics_pipeline")
              .cluster("prod")
              .jobId("aggregate_sales_metrics")
              .description("Aggregates sales data from multiple sources into summary tables")
              .name("Aggregate Sales Metrics")
              .type("BATCH")
              .build();

      // Add multiple input datasets (different sources)
      aggregationJob
          .addInputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.transactions,PROD)")
          .addInputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.customers,PROD)")
          .addInputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,raw.products,PROD)")
          .addInputDataset("urn:li:dataset:(urn:li:dataPlatform:kafka,events.purchases,PROD)");

      // Add multiple output datasets (different aggregation levels)
      aggregationJob
          .addOutputDataset(
              "urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.daily_sales,PROD)")
          .addOutputDataset(
              "urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.monthly_sales,PROD)")
          .addOutputDataset(
              "urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.customer_summary,PROD)");

      client.entities().upsert(aggregationJob);

      System.out.println("✓ Created aggregation job: " + aggregationJob.getUrn());
      System.out.println("  Input datasets: 4 (Snowflake tables + Kafka topic)");
      System.out.println("  Output datasets: 3 (daily, monthly, customer summaries)\n");

      // ==================== Example 3: Using setInputDatasets/setOutputDatasets
      // ====================
      System.out.println("Example 3: Setting multiple inlets/outlets at once");
      System.out.println("===================================================");

      DataJob batchJob =
          DataJob.builder()
              .orchestrator("dagster")
              .flowId("data_quality_pipeline")
              .cluster("prod")
              .jobId("validate_warehouse_tables")
              .description("Runs data quality checks on warehouse tables")
              .name("Validate Warehouse Tables")
              .type("BATCH")
              .build();

      // Set all inlets at once
      batchJob.setInputDatasets(
          Arrays.asList(
              "urn:li:dataset:(urn:li:dataPlatform:snowflake,dwh.orders,PROD)",
              "urn:li:dataset:(urn:li:dataPlatform:snowflake,dwh.customers,PROD)",
              "urn:li:dataset:(urn:li:dataPlatform:snowflake,dwh.products,PROD)"));

      // Set all outlets at once
      batchJob.setOutputDatasets(
          Arrays.asList(
              "urn:li:dataset:(urn:li:dataPlatform:snowflake,quality.validation_results,PROD)",
              "urn:li:dataset:(urn:li:dataPlatform:snowflake,quality.data_quality_metrics,PROD)"));

      client.entities().upsert(batchJob);

      System.out.println("✓ Created validation job: " + batchJob.getUrn());
      System.out.println("  Set 3 inlets and 2 outlets in batch operations\n");

      // ==================== Example 4: Updating Lineage ====================
      System.out.println("Example 4: Updating existing job lineage");
      System.out.println("=========================================");

      // Create a simple job first
      DataJob updateJob =
          DataJob.builder()
              .orchestrator("airflow")
              .flowId("data_processing_pipeline")
              .cluster("prod")
              .jobId("process_events")
              .description("Processes event data")
              .name("Process Events")
              .type("BATCH")
              .build();

      // Initial lineage
      updateJob.addInputDataset("urn:li:dataset:(urn:li:dataPlatform:kafka,events.raw,PROD)");
      updateJob.addOutputDataset("urn:li:dataset:(urn:li:dataPlatform:s3,processed/events,PROD)");

      client.entities().upsert(updateJob);
      System.out.println("✓ Created job with initial lineage");

      // Add another input source (e.g., requirement changed)
      updateJob.addInputDataset(
          "urn:li:dataset:(urn:li:dataPlatform:kafka,events.enrichment,PROD)");

      // Add another output destination
      updateJob.addOutputDataset(
          "urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.events,PROD)");

      client.entities().upsert(updateJob);
      System.out.println("✓ Updated job with additional inlet and outlet\n");

      // ==================== Example 5: Using DatasetUrn directly ====================
      System.out.println("Example 5: Using DatasetUrn objects for type safety");
      System.out.println("====================================================");

      DataJob typedJob =
          DataJob.builder()
              .orchestrator("spark")
              .flowId("ml_feature_pipeline")
              .cluster("prod")
              .jobId("generate_features")
              .description("Generates ML features from raw data")
              .name("Generate ML Features")
              .type("BATCH")
              .build();

      // Create DatasetUrn objects for type safety
      DatasetUrn sourceUrn =
          DatasetUrn.createFromString("urn:li:dataset:(urn:li:dataPlatform:hive,user_events,PROD)");
      DatasetUrn featureUrn =
          DatasetUrn.createFromString("urn:li:dataset:(urn:li:dataPlatform:hive,ml_features,PROD)");

      // Use typed URN objects
      typedJob.addInputDataset(sourceUrn).addOutputDataset(featureUrn);

      client.entities().upsert(typedJob);

      System.out.println("✓ Created job using DatasetUrn objects");
      System.out.println("  Type-safe URN creation prevents errors\n");

      // ==================== Example 6: Complete Data Pipeline ====================
      System.out.println("Example 6: Modeling a complete ETL pipeline");
      System.out.println("============================================");
      System.out.println("Pipeline: Extract → Transform → Load");
      System.out.println();

      // Extract job
      DataJob extractJob =
          DataJob.builder()
              .orchestrator("airflow")
              .flowId("complete_etl_pipeline")
              .cluster("prod")
              .jobId("extract_from_source")
              .description("Extracts data from operational database")
              .name("Extract from Source DB")
              .type("BATCH")
              .build();

      extractJob
          .addInputDataset("urn:li:dataset:(urn:li:dataPlatform:mysql,production.orders,PROD)")
          .addOutputDataset("urn:li:dataset:(urn:li:dataPlatform:s3,staging/orders_raw,PROD)");

      client.entities().upsert(extractJob);
      System.out.println("✓ Extract: mysql.orders → s3.staging/orders_raw");

      // Transform job
      DataJob transformJob =
          DataJob.builder()
              .orchestrator("airflow")
              .flowId("complete_etl_pipeline")
              .cluster("prod")
              .jobId("transform_data")
              .description("Cleanses and transforms extracted data")
              .name("Transform Data")
              .type("BATCH")
              .build();

      transformJob
          .addInputDataset("urn:li:dataset:(urn:li:dataPlatform:s3,staging/orders_raw,PROD)")
          .addOutputDataset("urn:li:dataset:(urn:li:dataPlatform:s3,staging/orders_clean,PROD)");

      client.entities().upsert(transformJob);
      System.out.println("✓ Transform: s3.staging/orders_raw → s3.staging/orders_clean");

      // Load job
      DataJob loadJob =
          DataJob.builder()
              .orchestrator("airflow")
              .flowId("complete_etl_pipeline")
              .cluster("prod")
              .jobId("load_to_warehouse")
              .description("Loads transformed data into data warehouse")
              .name("Load to Warehouse")
              .type("BATCH")
              .build();

      loadJob
          .addInputDataset("urn:li:dataset:(urn:li:dataPlatform:s3,staging/orders_clean,PROD)")
          .addOutputDataset("urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.orders,PROD)");

      client.entities().upsert(loadJob);
      System.out.println("✓ Load: s3.staging/orders_clean → snowflake.analytics.orders");
      System.out.println();
      System.out.println("Complete lineage created:");
      System.out.println(
          "  mysql.orders → [Extract] → s3.raw → [Transform] → s3.clean → [Load] → snowflake.analytics\n");

      // ==================== Summary ====================
      System.out.println("==========================================");
      System.out.println("Summary: Successfully created 9 data jobs with comprehensive lineage");
      System.out.println("==========================================");
      System.out.println("\nKey Takeaways:");
      System.out.println("  • Inlets represent input datasets (sources)");
      System.out.println("  • Outlets represent output datasets (destinations)");
      System.out.println("  • Use addInputDataset/addOutputDataset for single operations");
      System.out.println("  • Use setInputDatasets/setOutputDatasets for batch operations");
      System.out.println("  • DatasetUrn objects provide type safety");
      System.out.println("  • All jobs belong to a DataFlow (pipeline/DAG)");
      System.out.println("\nView your lineage graphs in DataHub UI!");

    } finally {
      client.close();
    }
  }
}

💡 Contributing to this documentation

This page is auto-generated from the underlying source code. To make changes, please edit the relevant source files in the metadata-integration directory.

Tip: For quick typo fixes or documentation updates, you can click the ✏️ Edit icon directly in the GitHub UI to open a Pull Request. For larger changes and PR naming conventions, please refer to our Contributing Guide.