使用 SIGKILL 耗时过长的气流终止任务

Question

我有一个 SQL Server 数据库，我将在其中以 parquet 格式迁移到 AWS S3 以构建一个数据湖。我正在使用 Apache Airflow 使用 DAGS 自动执行此任务。在这种情况下，架构上的每个表都成为一个 .parquet 文件，这使 S3 成为一个数据湖，因此能够使用 AWS Athena 后验和/或在 ElasticSearch 中进一步索引。

有一些非常大的表，这些表的迁移任务显然我希望花费更多时间。对于python，我发现唯一能连接微软SQL Server的库是pyodbc，它是微软官方开发和维护的。

对于如此大的表（6000 万个寄存器），使用 cursor.fetchall() 花费的时间太长并导致错误，因为该任务似乎被 Airlfow 通过 SIGNALKILL 杀死。

下面是 DAG 有多大的一个示例（只是其中的一部分）：

为了获取给定架构中的所有表，我使用以下 SQL Server 查询：

SELECT TABLE_NAME
FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_TYPE = 'BASE TABLE' AND TABLE_CATALOG='{}';

在括号中，我使用 Python 中的 .format() 函数插入模式名称并检索表以动态构建 DAG 结构。我已经更改了我的 python 代码以批量获取如此大的表中的数据，以尽量减少任何潜在的数据溢出：

def stream(cursor, batch_size=50000):
    while True:
        row = cursor.fetchmany(batch_size)
        if row is None or not row:
            break
        yield row


def fetch_data(query, schema, filename, remote_path, save_locally=False):
    cnxn = pyodbc.connect(driver='Here I Put the ODBC Driver Name',
                          host='Host for de SQL Server DB',
                          database='Nameof the DB Schema',
                          user='User for Auth in the DB',
                          password='Pass for Auth in the DB')
    print('Connetciton stabilished with {} ..'.format(schema))

    cursor = cnxn.cursor()
    print('Initializing cursor ...')
    print('Requestin query {} ..'.format(query))

    cursor.execute(query)
    print('Query fetched for {} ..'.format(schema))

    row_batch = stream(cursor)
    print('Getting Iterator ...')

    cols = cursor.description
    cols = [col[0] for col in cols]

    print('Creating batch data_frame ..')
    data_frame = pd.DataFrame(columns=cols)

    start_time = time.time()
    for rows in row_batch:
        batch_df = pd.DataFrame.from_records(rows, columns=cols)
        data_frame = data_frame.append(batch_df, ignore_index=True)
        batch_df = None
        print("-- Batch inserted in %s seconds --" % (time.time() - start_time))
        start_time = time.time()

    cnxn.close()
    print('Connetciton closed ..')
    
    // other code to convert to .parquet and send to S3
    save_to_bucket(data_frame, remote_path)
    return 'FETCHING DATA'

该策略似乎对模式的整个表的 96% 都有效，问题是，正如我之前所说，当表非常大，大约 6000 万条记录时，任务运行一段时间，大约30 分钟，但通常在那之后，Airflow 会终止任务，就像那样。没有连接错误，没有python异常或什么都没有。调度程序终端中唯一显示的是：

[2021-04-17 23:03:59,719] {scheduler_job.py:1199} INFO - Executor reports execution of ORTOCLIN_TO_S3.FETCHING_HISTORICORESUMO_DATA execution_date=2021-04-17 20:00:17.426578+00:00 exited with status success for try_number 1
[2021-04-17 23:05:02,050] {scheduler_job.py:1834} INFO - Resetting orphaned tasks for active dag runs
[2021-04-17 23:10:02,314] {scheduler_job.py:1834} INFO - Resetting orphaned tasks for active dag runs
[2021-04-17 23:15:02,666] {scheduler_job.py:1834} INFO - Resetting orphaned tasks for active dag runs
[2021-04-17 23:20:03,226] {scheduler_job.py:1834} INFO - Resetting orphaned tasks for active dag runs
[2021-04-17 23:25:03,868] {scheduler_job.py:1834} INFO - Resetting orphaned tasks for active dag runs
[2021-04-17 23:30:04,346] {scheduler_job.py:1834} INFO - Resetting orphaned tasks for active dag runs
[2021-04-17 23:35:04,853] {scheduler_job.py:1834} INFO - Resetting orphaned tasks for active dag runs
[2021-04-17 23:40:05,324] {scheduler_job.py:1834} INFO - Resetting orphaned tasks for active dag runs

在 Airflow 日志中，我们只需简单地执行以下任务：

有什么解决方法吗？请帮帮我！

Answer 1

您可以在此处更改 3 项内容：

AIRFLOW__CELERY__WORKER_CONCURRENCY

将此设置为较低的值或至少设置为 1，以使工作人员只专注于一项任务

AIRFLOW__CORE__KILLED_TASK_CLEANUP_TIME

将此设置为更高的值，例如 1200-3600 秒

增加worker机器的CPU和RAM

归根结底，这是一个资源问题。 DAG 消耗了太多资源并被杀死，所以这是一个合乎逻辑的步骤。

此列表不完整，可能还有其他解决方案，我还不知道。