我有一个 SQL Server 数据库,我正在其中以 parquet 格式迁移到 AWS S3 以构建数据湖。我正在使用 Apache Airflow 通过 DAGS 自动执行此任务。在这种情况下,模式上的每个表都变成了一个 .parquet 文件,这使 S3 成为一个数据湖,因此能够事后使用 AWS Athena 和/或在 ElasticSearch 中进一步编制索引。
有一些非常大的表,这些表的迁移任务显然我希望它花费更多的时间。对于python,我发现唯一与Microsoft SQL Server 连接的库是pyodbc,它是由Microsoft 官方开发和维护的。
对于如此大的表(大约 6000 万个寄存器),使用 cursor.fetchall()
花费的时间太长并导致错误,因为该任务似乎被 Airlfow 的 SIGNALKILL 杀死。
为了获取给定模式中的所有表,我使用了以下 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
有什么解决方法吗?请帮助我!
最佳答案
您可以在此处更改 3 项内容:
AIRFLOW__CELERY__WORKER_CONCURRENCY
将其设置为较低的值或至少设置为 1 以使工作人员只专注于一项任务
AIRFLOW__CORE__KILLED_TASK_CLEANUP_TIME
将其设置为更高的值,例如 1200-3600 秒
增加worker机器的CPU和RAM
归根结底这是资源问题。 DAG 消耗过多资源并被杀死,因此这是一个合乎逻辑的步骤。
此列表不完整,可能还有其他解决方案,我还不知道。
关于python - SIGKILL 耗时过长的 Airflow 终止任务,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/67143973/