如何在 apache beam GroupBy 中使用 datetime.date 值？

Question

我正在尝试使用 apache beam 的 beam.GroupBy(...) 构造，但遇到了不知道如何编码标准 python datetime.date 的问题。

这是一个演示该问题的简化代码块。代码：

import logging
import random
import apache_beam as beam
from apache_beam.io import WriteToText
from datetime import date


def random_record():
    return {
        'account_id'   : random.randint(1004, 1009),
        'activity_date': date(2021, 10, random.randint(1, 4)),
        'region'       : random.randint(30, 40),
        'largest_sale' : random.randint(10000, 40000),
        'total_sales'  : random.randint(100000, 900000)
    }

def main(argv=None):
    random.seed(2349090823434)

    records = [random_record() for i in range(100)]

    with beam.Pipeline() as p:
        output = (
            p 
            | "Source" >> beam.Create(records)
            | "Typed"  >> beam.Map(lambda d: beam.Row(**d))
            | "Rollup" >> (beam.GroupBy('account_id', 'activity_date')
                                .aggregate_field('total_sales', sum, 'total_sales')
                                .aggregate_field('largest_sale', max, 'largest_sale')
                        )
            | "Output" >> WriteToText('learn-output', file_name_suffix='.csv')
        )

if __name__ == '__main__':
    logging.getLogger().setLevel(logging.INFO)
    main()

当我运行这个时：

% python learn.py

导致错误：

WARNING:apache_beam.coders.coder_impl:Using fallback deterministic coder for type '<class 'apache_beam.transforms.core.Key'>' in 'Rollup/CombinePerKey(TupleCombineFn)/GroupByKey'.
Traceback (most recent call last):
  ...
TypeError: Unable to deterministically encode '2021-10-01' of type '<class 'datetime.date'>', please provide a type hint for the input of 'Rollup/CombinePerKey(TupleCombineFn)/GroupByKey'

...

TypeError: Unable to deterministically encode 'Key(account_id=1007, activity_date=datetime.date(2021, 10, 1))' of type '<class 'apache_beam.transforms.core.Key'>', 
please provide a type hint for the input of 'Rollup/CombinePerKey(TupleCombineFn)/GroupByKey'

我想知道如何实际“为 'Rollup/CombinePerKey(TupleCombineFn)/GroupByKey' 的输入提供类型提示”。

似乎在GroupBy 中调用了GroupByKey。所以我要么需要告诉 GroupBy 如何处理我的 2 元组键，要么注册一些编码器来支持 datetime.date...

Answer 1

通常你应该能够做类似的事情

class DateCoder(beam.coders.Coder):
    def encode(self, d):
        return d.isoformat().encode('ascii')
    def decode(self, bs):
        return datetime.date.fromisoformat(bs.decode('ascii'))
    def is_deterministic(self):
        return True

beam.coders.registry.register_coder(date, DateCoder)

来处理这个问题，但是这里与模式的交互很差，这使得在这种情况下使用beam.GroupBy 变得更加困难。一个问题是 beam.Row(**d) 实际上并没有让 Beam 在构造时找出这些 Row 对象的列名/类型（模式）（为此您需要显式传递关键字）。 （未来的工作会让这项工作变得更好。）

正如评论所建议的那样，绝对转换为字符串是一个简单的选项。

Answer 2

您可以将日期转换为字符串以进行分组：

beam.GroupBy('account_id', date_str=lambda x: str(x.activity_date))