3

我有一组同质字典，如何在不知道架构的情况下将它们写入 BigQuery？

BigQuerySink 要求我在构造它时指定架构。但是，我不知道架构：它是由我要编写的字典的键定义的。

有没有办法让我的管道推断架构，然后将其提供回（作为侧输入？）到接收器？

例如：

# Create a PCollection of dicts, something like
# {'field1': 'myval', 'field2': 10}
data = (p | 'generate_data' >> beam.ParDo(CreateData())

# Infer the schema from the data
# Generates a string for each element (ok to assume all dict keys equal)
# "field1:STRING, field2:INTEGER"
schema = (data
  | 'infer_schema' >> beam.ParDo(InferSchema())
  | 'sample_one'   >> beam.combiners.Sample.FixedSizeGlobally(1))

但是，我如何将架构作为参数提供给 BigQuerySink，并在 beam.io.Write 中使用它？

我知道这是不正确的，但我想做的是：

sink = BigQuerySink(tablename, dataset, project, schema=Materialize(schema))
p | 'write_bigquery' >> beam.io.Write(sink)

tl;dr 有没有办法从 apache Beam 以编程方式从数据中推断架构来创建和编写 bigquery 表？

4

2 回答 2

0

目前，我想出的最佳解决方案是明确硬编码 dict 键到 BigQuery 架构的映射。两个好处——它可以解决必须指定模式的问题，它可以让我从 BigQuery 中不想要的 dict 中过滤元素。

SCHEMA = {
  'field1': 'INTEGER',
  'field2': 'STRING',
  ...
}
schema_str = ','.join(['%s:%s' % (k, v) for k,v in SCHEMA.iteritems()])

sink = BigQuerySink(tablename,
        dataset=dataset,
        project=project,
        schema=schema_str,
        write_disposition=BigQueryDisposition.WRITE_TRUNCATE)

(pipeline
  # filters just the keys of each dict to the keys of SCHEMA.
  | 'filter_fields' >> beam.ParDo(FilterFieldKeysDoFn(SCHEMA))
  | 'to_bigquery' >> beam.io.Write(sink))
于 2017-07-07T15:57:28.973 回答
0

假设您的架构可以经常更改，那么将数据保持在更通用的形式可能会更好。

例如，您的行可能包含一条重复的记录（您的字典条目）。

记录模式如下所示： key (STRING) | 可选 string_val (STRING) | 可选的 int_val (INTEGER) 可选的 double_val (DOUBLE) | 可选 boolean_val (BOOLEAN) | ...

然后，您可以编写按类型扫描记录的查询。这效率有点低（因为如果它们位于不同的列中，您将不得不扫描您可能会跳过的行），但完全避免了预先指定您的模式。

于 2017-07-06T17:11:56.710 回答