Send task to queue in bulk - Celery Executor

[mik-laj]

Description

Hello,
I recently took care of CeleryExecutor. I managed to optimize the status retrieval by using bulk operations. Instead of fetching the status for each task using a separate query, one is sent for all tasks. This has accelerated this process more than 100 times in many cases.
#7542
However, we still use single requests in many processes to send tasks to the queue. This is very effective because of network latency.

airflow/airflow/executors/celery_executor.py

Lines 196 to 206 in f1dc2e0

	def _send_tasks_to_celery(self, task_tuples_to_send):
	# Use chunks instead of a work queue to reduce context switching
	# since tasks are roughly uniform in size
	chunksize = self._num_tasks_per_send_process(len(task_tuples_to_send))
	num_processes = min(len(task_tuples_to_send), self._sync_parallelism)
	with Pool(processes=num_processes) as send_pool:
	key_and_async_results = send_pool.map(
	send_task_to_executor,
	task_tuples_to_send,
	chunksize=chunksize)
	return key_and_async_results

It would be nice if it could be done as a bulk request in a single request. For Redis, this means using Pipeline.
https://github.com/andymccurdy/redis-py#pipelines

Can it be done easily in Celery?

Best regards,
Kamil

[mik-laj]

@auvipy Can you look at it? You're a Celery expert. I think Celery doesn't support it yet, but I might be wrong.

[auvipy]

celery redis need more care actually :) with my current time and other priorities in celery i didnt contribute much on redis part. I'm more focused on amqp 1.0 and kafka support and asyncio based worker....

[kurtqq]

this can be a good improvement