Custom Loop Extract | Smartchart开发手册

Introduction #

When a database contains time-suffixed partitioned tables (e.g., order_202001, order_202002) that need to be consolidated into a single target table, use the custom loop extraction feature. Combine the #diy driver with run_datax() for flexible multi-table loop sync.

Implementation Steps #

Step	Action
1	Create a DataX extraction task with parameter placeholders
2	Create a DIY task with loop logic
3	Orchestrate task dependencies in DAG

Usage #

SmartPip’s built-in datax component only supports single-table extraction without custom logic. Use the diy component for looping:

First, create a datax task with a parameter ZYM:

#datax job1  ZYM

Then create a diy task for loop extraction:

ZYM = '202001'
def fun_job2():
    job = os.path.join(ETL_FILE_PATH, 'project_name/job1.sql')
    report_time = datetime.date(2022, 8, 1)
    for i in range(600):
        zym = report_time.strftime('%Y%m')
        print(zym)
        if zym < '202001':
            break
        para_dict = {"ZYM": zym}
        run_datax(job, para_dict)
        report_time = (report_time - datetime.timedelta(days=1)).replace(day=1)

#diy  job2  fun_job2