Как передать аргументы в PythonOperator?

В PythonOperator позиционные аргументы передают через op_args, именованные — через op_kwargs. В python_callable указывают функцию, а не результат её вызова.

def process(source, limit):
    print(source, limit)

# Внутри определения DAG:
task = PythonOperator(
    task_id="process",
    python_callable=process,
    op_args=["events"],
    op_kwargs={"limit": 100},
)

Импорт PythonOperator зависит от установленной версии Airflow и провайдеров. Значения op_args и op_kwargs поддерживают шаблонизацию. Не передавайте через аргументы большие наборы данных: лучше передать путь или идентификатор и прочитать данные во время выполнения задачи.

Источник: Документация Apache Kafka.