Apache Airflow Celery Redis 解码错误答案

【问题标题】：Apache Airflow Celery Redis DecodeErrorApache Airflow Celery Redis 解码错误
【发布时间】：2017-09-19 12:16:20
【问题描述】：

使用最新版本的 apache 气流。从 LocalExecutor 开始，在该模式下一切正常，除了一些交互，Web UI 状态需要 CeleryExecutor 才能使用它们。使用 Redis 安装和配置 Celery 执行器，将 Redis 配置为代理 URL 和结果后端。

一开始它似乎可以工作，直到安排了一个任务，此时它给出了以下错误：

 File "/bin/airflow", line 28, in <module>
    args.func(args)
  File "/usr/lib/python2.7/site-packages/airflow/bin/cli.py", line 882, in scheduler
    job.run()
  File "/usr/lib/python2.7/site-packages/airflow/jobs.py", line 201, in run
    self._execute()
  File "/usr/lib/python2.7/site-packages/airflow/jobs.py", line 1311, in _execute
    self._execute_helper(processor_manager)
  File "/usr/lib/python2.7/site-packages/airflow/jobs.py", line 1444, in _execute_helper
    self.executor.heartbeat()
  File "/usr/lib/python2.7/site-packages/airflow/executors/base_executor.py", line 132, in heartbeat
    self.sync()
  File "/usr/lib/python2.7/site-packages/airflow/executors/celery_executor.py", line 91, in sync
    state = async.state
  File "/usr/lib/python2.7/site-packages/celery/result.py", line 436, in state
    return self._get_task_meta()['status']
  File "/usr/lib/python2.7/site-packages/celery/result.py", line 375, in _get_task_meta
    return self._maybe_set_cache(self.backend.get_task_meta(self.id))
  File "/usr/lib/python2.7/site-packages/celery/backends/base.py", line 352, in get_task_meta
    meta = self._get_task_meta_for(task_id)
  File "/usr/lib/python2.7/site-packages/celery/backends/base.py", line 668, in _get_task_meta_for
    return self.decode_result(meta)
  File "/usr/lib/python2.7/site-packages/celery/backends/base.py", line 271, in decode_result
    return self.meta_from_decoded(self.decode(payload))
  File "/usr/lib/python2.7/site-packages/celery/backends/base.py", line 278, in decode
    accept=self.accept)
  File "/usr/lib/python2.7/site-packages/kombu/serialization.py", line 263, in loads
    return decode(data)
  File "/usr/lib64/python2.7/contextlib.py", line 35, in __exit__
    self.gen.throw(type, value, traceback)
  File "/usr/lib/python2.7/site-packages/kombu/serialization.py", line 54, in _reraise_errors
    reraise(wrapper, wrapper(exc), sys.exc_info()[2])
  File "/usr/lib/python2.7/site-packages/kombu/serialization.py", line 50, in _reraise_errors
    yield
  File "/usr/lib/python2.7/site-packages/kombu/serialization.py", line 263, in loads
    return decode(data)
  File "/usr/lib/python2.7/site-packages/kombu/serialization.py", line 59, in pickle_loads
    return load(BytesIO(s))
kombu.exceptions.DecodeError: invalid load key, '{'.

似乎是pickle序列化错误，但我不确定如何追踪原因。有什么建议吗？

这个问题一直影响我使用 subdag 功能的工作流程，也许问题与此有关。

注意：我也使用rabbitMQ进行了测试，那里有不同的问题；客户端显示“对等方重置连接”并崩溃。 RabbitMQ 日志显示“客户端意外关闭 TCP 连接”。

【问题讨论】：

标签： airflow airflow-scheduler

【解决方案1】：

在我们的调度程序日志中看到完全相同的回溯后，我偶然发现了这一点：

  File "/usr/lib/python2.7/site-packages/kombu/serialization.py", line 59, in pickle_loads
    return load(BytesIO(s))
kombu.exceptions.DecodeError: invalid load key, '{'.

celery 试图解开以 '{' 开头的东西的事实似乎很可疑，因此我获取了流量的 tcpdump 并通过 Web UI 触发了一项任务。结果捕获包含此交换，几乎与上述回溯出现在调度程序日志中的同一时刻：

05:03:49.145849 IP <scheduler-ip-addr>.ec2.internal.45597 > <redis-ip-addr>.ec2.internal.6379: Flags [P.], seq 658:731, ack 46, win 211, options [nop,nop,TS val 654768546 ecr 4219564282], length 73: RESP "GET" "celery-task-meta-b0d3a29e-ac08-4e77-871e-b4d553502cc2"
05:03:49.146086 IP <redis-ip-addr>.ec2.internal.6379 > <scheduler-ip-addr>.ec2.internal.45597: Flags [P.], seq 46:177, ack 731, win 210, options [nop,nop,TS val 4219564282 ecr 654768546], length 131: RESP "{"status": "SUCCESS", "traceback": null, "result": null, "task_id": "b0d3a29e-ac08-4e77-871e-b4d553502cc2", "children": []}"

Redis 响应的有效载荷显然是 JSON，那么为什么 celery 试图解压呢？我们正在从 Airflow 1.7 迁移到 1.8，在我们推出期间，我们有一组 Airflow 工作人员运行 v1.7，另一组运行 v1.8。工作人员应该从具有不相交工作负载的队列中拉出，但由于我们的一个 DAG 中的错误，我们有一个由 Airflow 1.8 调度的 TaskInstance，然后由通过 Airflow 1.7 启动的 celery 工作人员执行。

AIRFLOW-1038 将 celery 任务状态的序列化程序从 JSON（默认）更改为 pickle，因此在此更改之前运行代码版本的工作人员将在 JSON 中序列化结果，调度程序运行的代码版本包括此更改将尝试通过 unpickling 反序列化结果，从而导致上述错误。

【讨论】：

【解决方案2】：

请验证您在airflow.cfg 中配置了哪种celery_result_backend。如果不是这样，请尝试将其切换到数据库后端（mysql 等）。

我们看到使用 ampq 后端（仅适用于 Celery 3.1 及更低版本），redis 和 rpc 后端有时会出现问题。

【讨论】：

【解决方案3】：

我们遇到了这个问题，我们的配置管理没有正确更新，因此调度程序和一些工作人员使用 apache-airflow 1.8.2，而大量工作人员正在运行气流 1.8.0。检查所有节点是否运行相同版本的气流。

【讨论】：