Hi I installed airflow in VirtualBox, while testing the below Code using the command airflow tasks test user_processing processing_user 2020-01-02 I am getting warning message and TaskNotFound error. (Followed Marc Lamberti course). Thanks in Advance.
My Code: created table using sqlite operator, used http operator to download the user from the API then used Python operator to process the information
from airflow.models import DAG
from datetime import datetime
from airflow.providers.sqlite.operators.sqlite import SqliteOperator
from airflow.providers.http.sensors.http import HttpSensor
from airflow.providers.http.operators.http import SimpleHttpOperator
import json
from airflow.operators.python import PythonOperator
from pandas import json_normalize
default_args = {'start_date' : datetime(2021,10,11)}
def _processing_user(ti):
users = ti.xcom_pull(task_id=['extracting_user'])
if not len(users) or 'results' not in users[0]:
raise ValueError('User is Empty')
user = users[0]['results'][0]
processed_user = json_normalize({
'firstname':user['name']['first'],
'lastname':user['name']['last'],
'country':user['location']['country'],
'username':user['login']['username'],
'password':user['login']['password'],
'email': user['email']
})
processed_user.to_csv('/tmp/processed_user.csv', index=None, header=False)
with DAG('user_processing',schedule_interval='@daily',default_args=default_args,catchup=False) as dag:
creating_table = SqliteOperator(
task_id='creating_table',
sqlite_conn_id='db_sqlite',
sql='''
CREATE TABLE users (
firstname TEXT NOT NULL,
lastname TEXT NOT NULL,
country TEXT NOT NULL,
password TEXT NOT NULL,
email TEXT NOT NULL PRIMARY KEY
);
'''
)
is_api_available = HttpSensor(
task_id='is_api_available',
http_conn_id='user_api',
endpoint='api/')
extracting_user = SimpleHttpOperator(
task_id='extracting_user',
http_conn_id='user_api',
endpoint='api/',
method='GET',
response_filter=lambda response: json.loads(response.text),
log_response=True
)
processing_user = PythonOperator(
task_id='Processing_user',
python_callable= _processing_user
)
Warning Message:
[2021-10-14 19:02:47,227] {dagbag.py:487} INFO - Filling up the DagBag from /home/airflow/airflow/dags
[2021-10-14 19:02:47,577] {baseoperator.py:1287} WARNING - Dependency <Task(_PythonDecoratedOperator): task_group_function.task_1>, task_group_function.task_2 already registered for DAG: example_task_group_decorator
[2021-10-14 19:02:47,579] {baseoperator.py:1287} WARNING - Dependency
[2021-10-14 19:02:47,586] {baseoperator.py:1287} WARNING - Dependency <Task(_PythonDecoratedOperator): task_group_function.task_1>, task_group_function.task_2 already registered for DAG: example_task_group_decorator
[2021-10-14 19:02:47,588] {baseoperator.py:1287} WARNING - Dependency
[2021-10-14 19:02:47,594] {baseoperator.py:1287} WARNING - Dependency
[2021-10-14 19:02:47,602] {baseoperator.py:1287} WARNING - Dependency <Task(_PythonDecoratedOperator): task_group_function.task_3>, task_group_function.task_2 already registered for DAG: example_task_group_decorator
[2021-10-14 19:02:47,604] {baseoperator.py:1287} WARNING - Dependency <Task(_PythonDecoratedOperator): task_group_function.task_2>, task_group_function.task_3 already registered for DAG: example_task_group_decorator
[2021-10-14 19:02:47,607] {baseoperator.py:1287} WARNING - Dependency <Task(_PythonDecoratedOperator): task_group_function.task_3>, task_group_function.task_2 already registered for DAG: example_task_group_decorator
[2021-10-14 19:02:47,609] {baseoperator.py:1287} WARNING - Dependency <Task(_PythonDecoratedOperator): task_group_function.task_2>, task_group_function.task_3 already registered for DAG: example_task_group_decorator
[2021-10-14 19:02:47,612] {baseoperator.py:1287} WARNING - Dependency <Task(_PythonDecoratedOperator): task_group_function.task_3>, task_group_function.task_2 already registered for DAG: example_task_group_decorator
[2021-10-14 19:02:47,615] {baseoperator.py:1287} WARNING - Dependency
Error Message:
Traceback (most recent call last):
File "/home/airflow/sandbox/bin/airflow", line 8, in <module>
sys.exit(main())
File "/home/airflow/sandbox/lib/python3.8/site-packages/airflow/__main__.py", line 40, in main
args.func(args)
File "/home/airflow/sandbox/lib/python3.8/site-packages/airflow/cli/cli_parser.py", line 48, in command
return func(*args, **kwargs)
File "/home/airflow/sandbox/lib/python3.8/site-packages/airflow/utils/cli.py", line 91, in wrapper
return f(*args, **kwargs)
File "/home/airflow/sandbox/lib/python3.8/site-packages/airflow/cli/commands/task_command.py", line 381, in task_test
task = dag.get_task(task_id=args.task_id)
File "/home/airflow/sandbox/lib/python3.8/site-packages/airflow/models/dag.py", line 1546, in get_task
raise TaskNotFound(f"Task {task_id} not found")
airflow.exceptions.TaskNotFound: Task processing_user not found