https://datahubproject.io logo
Join Slack
Powered by
Hello all, I'm impressed from the datahub software...
# ingestion
s
Hello all, I'm impressed from the datahub software so far to be honest. I try to get now also airflow 2.1.2 connected. I see the plugin in Admin->Plugins 
source	acryl-datahub-airflow-plugin==0.8.35.6: EntryPoint(name='acryl-datahub-airflow-plugin', value='datahub_airflow_plugin.datahub_plugin:DatahubPlugin', group='airflow.plugins')
I set also the lazy_loading to False. I added this in a test dag
Copy code
task2 = PythonOperator(
    task_id='Execute_Test_Script',
    python_callable=main,
    dag=dag,
    inlets={
      "datasets": [
        Dataset("postgres", "postgres.test.y"),
      ],
    },
    outlets={
      "datasets": [
        Dataset("postgres", "postgres.test.y"),
        Dataset("postgres", "postgres.test.x"),
        Dataset("postgres", "postgres.test.z")
      ]
    },
I added also the connection datahub_rest_default to the connections. But I don't see anything in the logs that it is emitting data. Has someone an idea what could be wrong?
c
s
@careful-pilot-86309 Thanks for the hint - I put exactly the dag as in the example but it does not send data out of airflow. I tested and the rest api is open to the docker image the plugin is installed. I don't see anything in the logs and no data is emmitted to datahub
d
@silly-ice-4153 the dataset array is not needed. This is how inlets/outlets should look like this:
Copy code
inlets=[
        Dataset("postgres", "postgres.test.y"),
    ],
    outlets=[
        Dataset("postgres", "postgres.test.y"),
        Dataset("postgres", "postgres.test.x"),
        Dataset("postgres", "postgres.test.z")
      ]
I see where it comes from, I think we need to update our examples
s
@dazzling-judge-80093 You are right I checked first the documentation and then was trying the code samples to verify that the installation works. I tried also the manual connection the thread is here https://datahubspace.slack.com/archives/CUMUWQU66/p1657898032644619 Is this example also not correct ?
d
you need to add the protocol in front of the url: http:// or https://
s
@dazzling-judge-80093 thanks for the answers - I changes my examples but I have the same result no error and no data in datahub. I have this in my airflow config
Copy code
[core]
core.lazy_load_plugins = false

[datahub]
graceful_exceptions = false
d
Can you show me how your inlets/outlets looks now? How do you run Airflow and which version?
s
I'm running Airflow 2.1.2 I'm running it in a docker container - I tested with a curl request to datahub and I get a response from there so the network is ok and I can reach it from the docker container.
Copy code
"""Lineage Backend
An example DAG demonstrating the usage of DataHub's Airflow lineage backend.
"""

from datetime import timedelta

from airflow import DAG
from airflow.utils.dates import days_ago

try:
    from airflow.operators.bash import BashOperator
except ModuleNotFoundError:
    from airflow.operators.bash_operator import BashOperator

from datahub_provider.entities import Dataset

default_args = {
    "owner": "airflow",
    "depends_on_past": False,
    "email": ["<mailto:jdoe@example.com|jdoe@example.com>"],
    "email_on_failure": False,
    "execution_timeout": timedelta(minutes=5),
}


with DAG(
    "datahub_lineage_backend_demo",
    default_args=default_args,
    description="An example DAG demonstrating the usage of DataHub's Airflow lineage backend.",
    schedule_interval=timedelta(days=1),
    start_date=days_ago(2),
    tags=["example_tag"],
    catchup=False,
) as dag:
    task1 = BashOperator(
        task_id="run_data_task",
        dag=dag,
        bash_command="echo 'This is where you might run your data tooling.'",
        inlets=[
            Dataset("postgres", "postgres.zoom.events"),
        ],
        outlets=[
            Dataset("postgres", "postgres.zoom.events"),
        ]
    )

task1
@dazzling-judge-80093 I modified the example to your suggestion
d
Can you see any datahub related log in your Task logs?
s
no I don't see anything there - the plugin I see as activated in the UI
Copy code
*** Reading local file: /opt/airflow/logs/datahub_lineage_backend_demo/run_data_task/2022-07-20T08:40:48.535939+00:00/1.log
[2022-07-20 08:40:49,721] {taskinstance.py:896} INFO - Dependencies all met for <TaskInstance: datahub_lineage_backend_demo.run_data_task 2022-07-20T08:40:48.535939+00:00 [queued]>
[2022-07-20 08:40:49,740] {taskinstance.py:896} INFO - Dependencies all met for <TaskInstance: datahub_lineage_backend_demo.run_data_task 2022-07-20T08:40:48.535939+00:00 [queued]>
[2022-07-20 08:40:49,740] {taskinstance.py:1087} INFO - 
--------------------------------------------------------------------------------
[2022-07-20 08:40:49,740] {taskinstance.py:1088} INFO - Starting attempt 1 of 1
[2022-07-20 08:40:49,740] {taskinstance.py:1089} INFO - 
--------------------------------------------------------------------------------
[2022-07-20 08:40:49,748] {taskinstance.py:1107} INFO - Executing <Task(BashOperator): run_data_task> on 2022-07-20T08:40:48.535939+00:00
[2022-07-20 08:40:49,754] {standard_task_runner.py:52} INFO - Started process 706 to run task
[2022-07-20 08:40:49,761] {standard_task_runner.py:76} INFO - Running: ['***', 'tasks', 'run', 'datahub_lineage_backend_demo', 'run_data_task', '2022-07-20T08:40:48.535939+00:00', '--job-id', '6423', '--pool', 'default_pool', '--raw', '--subdir', 'DAGS_FOLDER/lineage_test2.py', '--cfg-path', '/tmp/tmpje5rphow', '--error-file', '/tmp/tmp8s4zy_p_']
[2022-07-20 08:40:49,763] {standard_task_runner.py:77} INFO - Job 6423: Subtask run_data_task
[2022-07-20 08:40:49,835] {logging_mixin.py:104} INFO - Running <TaskInstance: datahub_lineage_backend_demo.run_data_task 2022-07-20T08:40:48.535939+00:00 [running]> on host efd30bf238e6
[2022-07-20 08:40:49,924] {taskinstance.py:1300} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_EMAIL=jdoe@example.com
AIRFLOW_CTX_DAG_OWNER=***
AIRFLOW_CTX_DAG_ID=datahub_lineage_backend_demo
AIRFLOW_CTX_TASK_ID=run_data_task
AIRFLOW_CTX_EXECUTION_DATE=2022-07-20T08:40:48.535939+00:00
AIRFLOW_CTX_DAG_RUN_ID=manual__2022-07-20T08:40:48.535939+00:00
[2022-07-20 08:40:49,940] {subprocess.py:52} INFO - Tmp dir root location: 
 /tmp
[2022-07-20 08:40:49,940] {subprocess.py:63} INFO - Running command: ['bash', '-c', "echo 'This is where you might run your data tooling.'"]
[2022-07-20 08:40:49,951] {subprocess.py:74} INFO - Output:
[2022-07-20 08:40:49,953] {subprocess.py:78} INFO - This is where you might run your data tooling.
[2022-07-20 08:40:49,953] {subprocess.py:82} INFO - Command exited with return code 0
[2022-07-20 08:40:50,046] {taskinstance.py:1204} INFO - Marking task as SUCCESS. dag_id=datahub_lineage_backend_demo, task_id=run_data_task, execution_date=20220720T084048, start_date=20220720T084049, end_date=20220720T084050
[2022-07-20 08:40:50,083] {taskinstance.py:1265} INFO - 0 downstream tasks scheduled from follow-on schedule check
[2022-07-20 08:40:50,091] {local_task_job.py:149} INFO - Task exited with return code 0
it succeeds but no emitting data lineage
d
Ahh, I see another issue. This is how your config should look like:
Copy code
[core]
lazy_load_plugins = false
you had an extra core bofore the lazy_load_plugins which does not need because the 
[core]
defines it
s
ahh too much searching around - I put it like that in and the airflow config said that it needs a header that's why I put core inside - seems that I forgot to delete the core before. Thanks so much that did it It is working now
d
awesome
s
One fast question Do I have to define inlets because the sources come from outside system I don't have a reach ?
d
If you want that to show up in Datahub and see in the lineage then yes, even thought it will be just a virtual node
s
ok so virtual nodes are possible even if the dataset is not in datahub directly ?
d
yes, it is but it won’t have schema and others
s
ok thanks so much for the information in the long run I try to get internal budget to contribute to datahub because we would need more ingestion sources
d
That would be awesome, thanks. Which sources are you missing?
s
I have to test the salesforce connector - We need that and some special applications like hubspot, I have to check about atlassian if this is useful.
Mostly API related sources
4 Views
Open in Slack
PreviousNext
Powered by