p_utils = imp.load_source(u’p_utils’, join(’/efs/com/cm/utilities/cm_integration_python_utils.py’))
lash_site_init_task = PythonOperator(task_id=‘xyz’,python_callable=p_utils.initiate_batch, op_kwargs={‘source’:‘lash’,‘feed’:‘site’}, trigger_rule=TriggerRule.ALL_SUCCESS,dag=lash_dag)
----------------------------------------------------here P_utils is utility in python and initiate_batch --------
I want to convert pythonoperator to k8podoperator
But i am facing issue not working there need correct way to pass utility.function to k8podoperator
tried below not works
using k8podopearator
lash_site_init_task = KubernetesPodOperator(namespace=namespace, image=image , cmds=["/bin/bash","-c", p_utils.compare_lst_modified_time], op_kwargs={‘source’:‘lash’,‘feed’:‘prvdr’}, volume_mounts=[volume_mount], volumes=[volume], name=“table-lash_prvdr_data_load_check”, task_id=‘lash_prvdr_data_load_check’, get_logs=True, provide_context=True, config_file=kube_config_path, in_cluster=False, cluster_context=‘aws’, trigger_rule=TriggerRule.ALL_SUCCESS, on_failure_callback=p_utils.update_audit_failed_status(‘lash’, ‘prvdr’, ‘data_load_check’), dag=lash_dag)