对表进行分区

Question

Bigquery 目前只允许按日期进行分区。

假设我有一个包含 inserted_timestamp 字段的 10 亿表行。假设该字段的日期为 1 年前。

将现有数据移动到新分区表的正确方法是什么？

已编辑

我看到有一个关于 Java 版本 Sharding BigQuery output tables 的优雅解决方案也在 BigQuery partitioning with Beam streams 详细说明，即参数化表名（或分区后缀）窗口数据。

但我在 2.x 梁项目上想念BigQueryIO.Write，也没有关于从 python 可序列化函数获取窗口时间的示例。

我尝试在管道上创建分区，但如果因大量分区而失败（以 100 运行但因 1000 失败）。

这是我的代码：

               (  p
                | 'lectura' >> beam.io.ReadFromText(input_table)
                | 'noheaders' >> beam.Filter(lambda s: s[0].isdigit())
                | 'addtimestamp' >> beam.ParDo(AddTimestampDoFn())
                | 'window' >> beam.WindowInto(beam.window.FixedWindows(60))
                | 'table2row'  >> beam.Map( to_table_row )  
                | 'write2table' >> beam.io.Write(beam.io.BigQuerySink(
                        output_table,   #<-- unable to parametrize by window
                        dataset=my_dataset, 
                        project=project, 
                        schema='dia:DATE, classe:STRING, cp:STRING, import:FLOAT',
                        create_disposition=CREATE_IF_NEEDED,
                        write_disposition=WRITE_TRUNCATE,
                                    )
                                )
                )

p.run()

Answer 1

所有必要的功能都存在于 Beam 中，但目前可能仅限于 Java SDK。

您将使用BigQueryIO。具体来说，您可以使用DynamicDestinations 来确定每一行的目标表。

以 DynamicDestinations 为例：

events.apply(BigQueryIO.<UserEvent>write()
  .to(new DynamicDestinations<UserEvent, String>() {
        public String getDestination(ValueInSingleWindow<String> element) {
          return element.getValue().getUserId();
        }
        public TableDestination getTable(String user) {
          return new TableDestination(tableForUser(user), 
            "Table for user " + user);
        }
        public TableSchema getSchema(String user) {
          return tableSchemaForUser(user);
        }
      })
  .withFormatFunction(new SerializableFunction<UserEvent, TableRow>() {
     public TableRow apply(UserEvent event) {
       return convertUserEventToTableRow(event);
     }
   }));