使用 DataflowRunner 时无法将 Apache Beam 写入输出到 bigquery

Question

我试图理解为什么这个管道不会向BigQuery 写入任何输出。 我想要实现的是从不同的货币对观察开始计算过去 10 年的美元指数。

所有数据都在BigQuery 中，我需要按时间顺序对其进行组织和排序（如果有更好的方法来实现这一点，我很高兴阅读它，因为我认为这可能不是最佳方法）。

Currencies() 类的想法是开始对货币对（例如：EURUSD）的最后一次观察进行分组（并保留），在所有货币对“到达”时更新它们，按时间顺序对它们进行排序，最后得到当日美元指数的开盘价、最高价、最低价和收盘价。

此代码在我的 jupyter 笔记本和使用 DirectRunner 的云 shell 中有效，但是当我使用 DataflowRunner 时，它不会写入任何输出。为了看看我是否能弄明白，我尝试使用beam.Create() 创建数据，然后将其写入BigQuery（它有效），还只是从BQ 读取一些内容并将其写入其他表（也有效），所以我最好的猜测是问题出在beam.CombineGlobally 部分，但我不知道它是什么。

代码如下：

import logging
import collections
import apache_beam as beam

from datetime import datetime

SYMBOLS = ['usdjpy', 'usdcad', 'usdchf', 'eurusd', 'audusd', 'nzdusd', 'gbpusd']

TABLE_SCHEMA = "date:DATETIME,index:STRING,open:FLOAT,high:FLOAT,low:FLOAT,close:FLOAT"

class Currencies(beam.CombineFn):
    def create_accumulator(self):
        return {}
    
    def add_input(self,accumulator,inputs):
        logging.info(inputs)
        date,currency,bid = inputs.values()
        if '.' not in date:
            date = date+'.0'        
        date = datetime.strptime(date,'%Y-%m-%dT%H:%M:%S.%f')
        data = currency+':'+str(bid)
        accumulator[date] = [data]
        return accumulator
    
    def merge_accumulators(self,accumulators):
        merged = {}
        for accum in accumulators:
            ordered_data = collections.OrderedDict(sorted(accum.items()))
            prev_date = None
            for date,date_data in ordered_data.items():
                if date not in merged:
                    merged[date] = {}
                    if prev_date is None:
                        prev_date = date
                    else:
                        prev_data = merged[prev_date]
                        merged[date].update(prev_data)
                        prev_date = date

                for data in date_data:
                    currency,bid = data.split(':')
                    bid = float(bid)
                    currency = currency.lower()
                    merged[date].update({
                        currency:bid
                    })
        return merged
    
    def calculate_index_value(self,data):
        return data['usdjpy']*data['usdcad']*data['usdchf']/(data['eurusd']*data['audusd']*data['nzdusd']*data['gbpusd'])
    
    def extract_output(self,accumulator):
        ordered = collections.OrderedDict(sorted(accumulator.items()))
        index = {}
        for dt,currencies in ordered.items():
            if not all([symbol in currencies.keys() for symbol in SYMBOLS]):
                continue
            date = str(dt.date())
            index_value = self.calculate_index_value(currencies)
            if date not in index:
                index[date] = {
                    'date':date,
                    'index':'usd',
                    'open':index_value,
                    'high':index_value,
                    'low':index_value,
                    'close':index_value
                }
            else:
                max_value = max(index_value,index[date]['high'])
                min_value = min(index_value,index[date]['low'])
                close_value = index_value
                index[date].update({
                    'high':max_value,
                    'low':min_value,
                    'close':close_value
                })
        return index


def main():
    query = """
    select date,currency,bid from data_table
    where date(date) between '2022-01-13' and '2022-01-16'
    and currency like ('%USD%')
    """

    options = beam.options.pipeline_options.PipelineOptions(
        temp_location = 'gs://PROJECT/temp',
        project = 'PROJECT',
        runner = 'DataflowRunner',
        region = 'REGION',
        num_workers = 1,
        max_num_workers = 1,
        machine_type = 'n1-standard-1',
        save_main_session = True,
        staging_location = 'gs://PROJECT/stag'
    )
    with beam.Pipeline(options = options) as pipeline:
        inputs = (pipeline
                 | 'Read From BQ' >> beam.io.ReadFromBigQuery(query=query,use_standard_sql=True)
                 | 'Accumulate' >> beam.CombineGlobally(Currencies())
                 | 'Flat' >> beam.ParDo(lambda x: x.values())
                 | beam.io.Write(beam.io.WriteToBigQuery(
                     table = 'TABLE',
                     dataset = 'DATASET',
                     project = 'PROJECT',
                     schema = TABLE_SCHEMA))
                 )
        
if __name__ == '__main__':
    logging.getLogger().setLevel(logging.INFO)
    main()

我执行此操作的方式是从 shell，使用 python3 -m first_script（这是我应该运行此批处理作业的方式吗？）。 我错过了什么或做错了什么？这是我第一次尝试使用 Dataflow，所以我可能在书中犯了几个错误。

Answer 1

它可能对谁有帮助：我遇到了类似的问题，但我已经为不同的流程使用了相同的代码，该流程有一个 pubsub 作为输入，它可以完美地工作，而不是基于文件的输入，而它根本没有。经过大量的实验，我发现在选项中我改变了标志

options = PipelineOptions(streaming=True, ..

到

options = PipelineOptions(streaming=False,

当然，它不是流式源，而是有界源，批处理。将此标志设置为 true 后，我在 BigQuery 表中找到了我的行。完成后，它甚至停止了管道，因为它在哪里进行批处理操作。希望这会有所帮助