如何使用 Python Apache Beam 解压缩 beam.Map 中的字典值

Question

我在 beam.Map 中使用了这个函数：

def to_row(self, record):
    tags = {
        "foo": "bar"
    }

    x = {**tags}

    return beam.Row(
        date=record.get("date"),
        value=record.get("input_value").get("value"),
    )

当函数在管道中运行时，由于语句 x={**tags}，它会中断并给出以下错误：

 TypeError: Attempted to determine schema for unsupported type 'Any'

如果我删除 x=... 语句，一切正常。

我知道 Apache Beam 是在 python 中输入的，但我真的不明白为什么我不能解压值。

有没有办法在没有这个问题的情况下正确解包？理想情况下，我想将 Dict[str,str] 传递给 beam.Row 没有任何问题。

Answer 1

x 和 tags 在您的代码中使用在哪里？

如果您想将Dict[str, str] 传递给beam.Row，只需解压即可：beam.Row(**tags)。

在您的代码中，x 仍然是等于 tags 的 Dict[str, str]。