Apache airflow: Cannot use more than 1 thread when using sqlite. Setting max_threads to 1

Viewed 4966

I have setup airflow with postgresql database, I am creating multiple dags

def subdag(parent_dag_name, child_dag_name,currentDate,batchId,category,subCategory,yearMonth,utilityType,REALTIME_HOME, args):
dag_subdag = DAG(
    dag_id='%s.%s' % (parent_dag_name, child_dag_name),
    default_args=args,
    schedule_interval="@once",
)
# get site list to run bs reports
site_list = getSiteListforProcessing(category,subCategory,utilityType,yearMonth);
print (site_list)
def update_status(siteId,**kwargs):
    createdDate=getCurrentTimestamp();
    print ('N',siteId,batchId,yearMonth,utilityType,'N')
    updateJobStatusLog('N',siteId,batchId,yearMonth,utilityType,'P')

def error_status(siteId,**kwargs):
    createdDate=getCurrentTimestamp();
    print ('N',siteId,batchId,yearMonth,utilityType,'N')

BS_template = """
echo "{{ params.date }}"
java -cp xx.jar com.xGenerator {{params.siteId}} {{params.utilityType}} {{params.date}}
"""

for index,siteid in enumerate(site_list):
    t1 = BashOperator(
            task_id='%s-task-%s' % (child_dag_name, index + 1),
            bash_command=BS_template,
            params={'date':  currentDate, 'realtime_home': REALTIME_HOME,'siteId': siteid, "utilityType":utilityType},
            default_args=args,
            dag=dag_subdag)
    t2 = PythonOperator(
            task_id='%s-updatetask-%s' % (child_dag_name, index + 1),
            dag=dag_subdag,
            python_callable=update_status,
            op_kwargs={'siteId':siteid})

    t2.set_upstream(t1)
return dag_subdag

It creates dynamic tasks but on all number of dynamic task, it always fails the last one and logs error as: "Cannot use more than 1 thread when using sqlite. Setting max_threads to 1" E.g. : if 4 tasks are created 3 runs, and if 2 tasks are created 1 runs.

0 Answers
Related