{"library":"openlineage-python","code":"import os\nfrom datetime import datetime\nimport uuid\n\nfrom openlineage.client.client import OpenLineageClient\nfrom openlineage.client.event_v2 import RunEvent, RunState, Job, InputDataset, OutputDataset, Run\n\n# Configure OpenLineage to send events to the console for demonstration\nos.environ['OPENLINEAGE_URL'] = os.environ.get('OPENLINEAGE_URL', 'console') # Use 'console' for local output\nos.environ['OPENLINEAGE_NAMESPACE'] = os.environ.get('OPENLINEAGE_NAMESPACE', 'my_app_namespace')\n\n# Initialize the OpenLineage client\nclient = OpenLineageClient()\n\ndef my_data_processing_job():\n    job_name = \"my_simple_job\"\n    run_id = str(uuid.uuid4())\n    namespace = os.environ['OPENLINEAGE_NAMESPACE']\n    \n    input_dataset_name = \"input_data\"\n    output_dataset_name = \"processed_data\"\n\n    # 1. Emit START event\n    start_event = RunEvent(\n        eventType=RunState.START,\n        eventTime=datetime.now().isoformat(),\n        run=Run(runId=run_id, facets={}),\n        job=Job(namespace=namespace, name=job_name, facets={}),\n        inputs=[InputDataset(namespace=namespace, name=input_dataset_name)],\n        outputs=[OutputDataset(namespace=namespace, name=output_dataset_name)],\n        producer=client.producer,\n        schemaURL=client.schema_url_v2\n    )\n    client.emit(start_event)\n    print(f\"Emitted START event for job '{job_name}' with run ID '{run_id}'\")\n\n    try:\n        # Simulate data processing\n        print(f\"Processing data for job '{job_name}'...\")\n        # Add actual processing logic here\n\n        # 2. Emit COMPLETE event on success\n        complete_event = RunEvent(\n            eventType=RunState.COMPLETE,\n            eventTime=datetime.now().isoformat(),\n            run=Run(runId=run_id, facets={}),\n            job=Job(namespace=namespace, name=job_name, facets={}),\n            inputs=[InputDataset(namespace=namespace, name=input_dataset_name)],\n            outputs=[OutputDataset(namespace=namespace, name=output_dataset_name)],\n            producer=client.producer,\n            schemaURL=client.schema_url_v2\n        )\n        client.emit(complete_event)\n        print(f\"Emitted COMPLETE event for job '{job_name}'\")\n\n    except Exception as e:\n        print(f\"Job '{job_name}' failed: {e}\")\n        # 3. Emit FAIL event on failure\n        fail_event = RunEvent(\n            eventType=RunState.FAIL,\n            eventTime=datetime.now().isoformat(),\n            run=Run(runId=run_id, facets={}),\n            job=Job(namespace=namespace, name=job_name, facets={}),\n            inputs=[InputDataset(namespace=namespace, name=input_dataset_name)],\n            outputs=[OutputDataset(namespace=namespace, name=output_dataset_name)],\n            producer=client.producer,\n            schemaURL=client.schema_url_v2\n        )\n        client.emit(fail_event)\n        print(f\"Emitted FAIL event for job '{job_name}'\")\n\nif __name__ == \"__main__\":\n    my_data_processing_job()","lang":"python","description":"This quickstart demonstrates how to initialize the `OpenLineageClient` and manually emit `START` and `COMPLETE` (or `FAIL`) events for a data processing job. It sets the `OPENLINEAGE_URL` to 'console' to print events directly to standard output, making it easy to see the generated lineage without a full OpenLineage backend.","tag":null,"tag_description":null,"last_tested":"2026-04-24","results":[{"runtime":"python:3.10-alpine","exit_code":1},{"runtime":"python:3.10-slim","exit_code":1},{"runtime":"python:3.11-alpine","exit_code":1},{"runtime":"python:3.11-slim","exit_code":1},{"runtime":"python:3.12-alpine","exit_code":1},{"runtime":"python:3.12-slim","exit_code":1},{"runtime":"python:3.13-alpine","exit_code":1},{"runtime":"python:3.13-slim","exit_code":1},{"runtime":"python:3.9-alpine","exit_code":1},{"runtime":"python:3.9-slim","exit_code":1}]}