Skip to content

Instantly share code, notes, and snippets.

@pingzh

pingzh/0_repeated_mapping.py

Last active May 31, 2022 22:52
Show Gist options
  • Save pingzh/b9abaeac2357a496d0675acde439b736 to your computer and use it in GitHub Desktop.
Save pingzh/b9abaeac2357a496d0675acde439b736 to your computer and use it in GitHub Desktop.
Download ZIP
test dag for 2.3.2rc1
Raw
0_repeated_mapping.py
import csv
import io
import os
import json
from datetime import datetime
from airflow import DAG
from airflow.decorators import task
with DAG(dag_id="0_repeated_mapping", start_date=datetime(2022, 3, 4)) as dag:
@task
def add_one(x: int):
return x + 1
first = add_one.expand(x=[1, 2, 3])
second = add_one.expand(x=first)
@pingzh
Copy link
Author

pingzh commented May 31, 2022 •
edited
Loading

env:

❯ grep sql_alchemy_conn $AIRFLOW_HOME/airflow.cfg
sql_alchemy_conn = mysql://airflow:airflow@host.docker.internal/airflow_apache
# sql_alchemy_connect_args =
sql_alchemy_conn = mysql://airflow:airflow@host.docker.internal/airflow_apache

❯ python -V
Python 3.9.7

OS: Mac M1:

❯ uname  -a
Darwin Pings-MBP 21.5.0 Darwin Kernel Version 21.5.0: Tue Apr 26 21:08:37 PDT 2022; root:xnu-8020.121.3~4/RELEASE_ARM64_T6000 arm64

mysql version:

❯ mysql -V
mysql  Ver 8.0.28 for macos12.0 on arm64 (Homebrew)

set up:

I downloaded the tar.gz and installed it pip install /Users/ping_zhang/Desktop/apache-airflow-2.3.2.tar.gz

mysql -u root -e 'drop database airflow_apache; create database airflow_apache; '

airflow users create --username admin --firstname Demo  --lastname Demo --role Admin --email spiderman@superhero.org -p 123456

# put the above dag under the dags_folder

start webserver, scheduler and worker

error in the grid view

Python version: 3.9.7
Airflow version: 2.3.2
Node: 11.1.168.192.in-addr.arpa
-------------------------------------------------------------------------------
Traceback (most recent call last):
  File "/Users/ping_zhang/miniforge3/envs/apache-airflow/lib/python3.9/site-packages/flask/app.py", line 2447, in wsgi_app
    response = self.full_dispatch_request()
  File "/Users/ping_zhang/miniforge3/envs/apache-airflow/lib/python3.9/site-packages/flask/app.py", line 1952, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/Users/ping_zhang/miniforge3/envs/apache-airflow/lib/python3.9/site-packages/flask/app.py", line 1821, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "/Users/ping_zhang/miniforge3/envs/apache-airflow/lib/python3.9/site-packages/flask/_compat.py", line 39, in reraise
    raise value
  File "/Users/ping_zhang/miniforge3/envs/apache-airflow/lib/python3.9/site-packages/flask/app.py", line 1950, in full_dispatch_request
    rv = self.dispatch_request()
  File "/Users/ping_zhang/miniforge3/envs/apache-airflow/lib/python3.9/site-packages/flask/app.py", line 1936, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "/Users/ping_zhang/miniforge3/envs/apache-airflow/lib/python3.9/site-packages/airflow/www/auth.py", line 43, in decorated
    return func(*args, **kwargs)
  File "/Users/ping_zhang/miniforge3/envs/apache-airflow/lib/python3.9/site-packages/airflow/www/decorators.py", line 117, in view_func
    return f(*args, **kwargs)
  File "/Users/ping_zhang/miniforge3/envs/apache-airflow/lib/python3.9/site-packages/airflow/www/decorators.py", line 80, in wrapper
    return f(*args, **kwargs)
  File "/Users/ping_zhang/miniforge3/envs/apache-airflow/lib/python3.9/site-packages/airflow/utils/session.py", line 71, in wrapper
    return func(*args, session=session, **kwargs)
  File "/Users/ping_zhang/miniforge3/envs/apache-airflow/lib/python3.9/site-packages/airflow/www/views.py", line 2635, in grid
    'groups': task_group_to_grid(dag.task_group, dag, dag_runs, session),
  File "/Users/ping_zhang/miniforge3/envs/apache-airflow/lib/python3.9/site-packages/airflow/www/views.py", line 270, in task_group_to_grid
    children = [task_group_to_grid(child, dag, dag_runs, session) for child in task_group.topological_sort()]
  File "/Users/ping_zhang/miniforge3/envs/apache-airflow/lib/python3.9/site-packages/airflow/www/views.py", line 270, in <listcomp>
    children = [task_group_to_grid(child, dag, dag_runs, session) for child in task_group.topological_sort()]
  File "/Users/ping_zhang/miniforge3/envs/apache-airflow/lib/python3.9/site-packages/airflow/www/views.py", line 261, in task_group_to_grid
    'instances': wwwutils.get_task_summaries(task_item_or_group, dag_runs, session),
  File "/Users/ping_zhang/miniforge3/envs/apache-airflow/lib/python3.9/site-packages/airflow/www/utils.py", line 170, in get_task_summaries
    return [_get_summary(ti) for ti in tis]
  File "/Users/ping_zhang/miniforge3/envs/apache-airflow/lib/python3.9/site-packages/airflow/www/utils.py", line 170, in <listcomp>
    return [_get_summary(ti) for ti in tis]
  File "/Users/ping_zhang/miniforge3/envs/apache-airflow/lib/python3.9/site-packages/airflow/www/utils.py", line 151, in _get_summary
    task_instance, task_instances=get_mapped_instances(task_instance, session)
  File "/Users/ping_zhang/miniforge3/envs/apache-airflow/lib/python3.9/site-packages/airflow/www/utils.py", line 64, in get_mapped_instances
    TaskInstance.dag_id == task_instance.dag_id,
AttributeError: Could not locate column in row for column 'dag_id'

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment