ElasticSearch 的 Celery 结果类型答案

【问题标题】：Celery Result type for ElasticSearchElasticSearch 的 Celery 结果类型
【发布时间】：2019-12-02 22:28:47
【问题描述】：

我目前正在为我的工作探索 celery，并且我正在尝试设置 Elasticsearch 后端。有没有办法将结果值作为字典/JSON而不是文本发送？因此，Elasticsearch 中的结果会正确显示并且可以使用嵌套类型吗？

celery 创建的自动映射

{
  "celery" : {
    "mappings" : {
      "backend" : {
        "properties" : {
          "@timestamp" : {
            "type" : "date"
          },
          "result" : {
            "type" : "text",
            "fields" : {
              "keyword" : {
                "type" : "keyword",
                "ignore_above" : 256
              }
            }
          }
        }
      }
    }
  }
}

我尝试使用嵌套字段创建自己的映射，但结果是 elasticsearch.exceptions.RequestError: RequestError(400, 'mapper_parsing_exception', 'object mapping for [result] tried to parse field [result] as object, but found a concrete value')

更新

结果已经用 JSON 编码，并且在 Elasticsearch 包装器中 JSON 字符串保存在字典中。添加json.loads(result) 作为快速修复实际上会有所帮助。

快速修复新映射出现后：

{
  "celery" : {
    "mappings" : {
      "backend" : {
        "properties" : {
          "@timestamp" : {
            "type" : "date"
          },
          "result" : {
            "properties" : {
              "date_done" : {
                "type" : "date"
              },
              "result" : {
                "type" : "long"
              },
              "status" : {
                "type" : "text",
                "fields" : {
                  "keyword" : {
                    "type" : "keyword",
                    "ignore_above" : 256
                  }
                }
              },
              "task_id" : {
                "type" : "text",
                "fields" : {
                  "keyword" : {
                    "type" : "keyword",
                    "ignore_above" : 256
                  }
                }
              }
            }
          }
        }
      }
    }
  }
}

更新了 Kibana 视图：

有什么方法可以在 Celery 中禁用结果序列化？

我可以添加一个带有解包 JSON 的拉取请求，仅用于 Elasticsearch，但它看起来像一个 hack。

【问题讨论】：

标签： python json elasticsearch celery

【解决方案1】：

从 v4.0 开始，默认的 result_serializer 是 json，所以无论如何你都应该得到 JSON 格式的结果。也许您的配置使用了其他东西？ - 在这种情况下，我建议你删除它（如果你使用 Celery >=4.0），你应该享受 JSON 格式的结果。我更喜欢 msgpack 但另一方面我不在 Celery 结果上使用 ElasticSearch...

【讨论】：

我刚刚安装了 Celery (4.3.0) 并且没有更改任何配置。映射是自动创建的。我能看到的唯一问题是在_indexfunction 中的elasticsearch 的芹菜代码中，结果被转换为字典，但后来没有转换为JSON，因此elasticsearch 认为它只是一个文本。目前我能看到的唯一解决方案是 fork celery 并添加一行 json.dumps({bytes_to_str(k): v for k, v in items(body)})，但这不是最好的解决方案。
您是否尝试将序列化格式显式设置为 JSON？
Celery 文档非常混乱，我尝试添加所有提及 json app = Celery(backend='elasticsearch://..., accept_content=['json'], result_accept_content = ['json'], serializer='json') 结果仍然是文本
你指定了 result_serializer 吗？
刚刚添加了result_serializer = 'json'。反正也没用。