Создание / запись в разделенную таблицу BigQuery через поток данных Google Cloud

Как говорит Паван, запись в таблицы разделов с помощью Dataflow определенно возможна. Вы используете DataflowPipelineRunner в потоковом или пакетном режиме?

Предложенное вами решение должно работать. В частности, если вы предварительно создаете таблицу с настроенным разделением по дате, вы можете использовать BigQueryIO.Write.toTableReference лямбда для записи в раздел по дате. Например:

/**
 * A Joda-time formatter that prints a date in format like {@code "20160101"}.
 * Threadsafe.
 */
private static final DateTimeFormatter FORMATTER =
    DateTimeFormat.forPattern("yyyyMMdd").withZone(DateTimeZone.UTC);

// This code generates a valid BigQuery partition name:
Instant instant = Instant.now(); // any Joda instant in a reasonable time range
String baseTableName = "project:dataset.table"; // a valid BigQuery table name
String partitionName =
    String.format("%s$%s", baseTableName, FORMATTER.print(instant));

Подход, который я использовал (работает и в потоковом режиме):

• Определите настраиваемое окно для входящей записи

• Преобразуйте окно в название таблицы / раздела

  p.apply(PubsubIO.Read
              .subscription(subscription)
              .withCoder(TableRowJsonCoder.of())
          )
          .apply(Window.into(new TablePartitionWindowFn()) )
          .apply(BigQueryIO.Write
                         .to(new DayPartitionFunc(dataset, table))
                         .withSchema(schema)
                         .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
          );
  

Установив окно на основе входящих данных, End Instant можно игнорировать, так как начальное значение используется для настройки раздела:

public class TablePartitionWindowFn extends NonMergingWindowFn<Object, IntervalWindow> {

private IntervalWindow assignWindow(AssignContext context) {
    TableRow source = (TableRow) context.element();
    String dttm_str = (String) source.get("DTTM");

    DateTimeFormatter formatter = DateTimeFormat.forPattern("yyyy-MM-dd").withZoneUTC();

    Instant start_point = Instant.parse(dttm_str,formatter);
    Instant end_point = start_point.withDurationAdded(1000, 1);

    return new IntervalWindow(start_point, end_point);
};

@Override
public Coder<IntervalWindow> windowCoder() {
    return IntervalWindow.getCoder();
}

@Override
public Collection<IntervalWindow> assignWindows(AssignContext c) throws Exception {
    return Arrays.asList(assignWindow(c));
}

@Override
public boolean isCompatible(WindowFn<?, ?> other) {
    return false;
}

@Override
public IntervalWindow getSideInputWindow(BoundedWindow window) {
    if (window instanceof GlobalWindow) {
        throw new IllegalArgumentException(
                "Attempted to get side input window for GlobalWindow from non-global WindowFn");
    }
    return null;
}

Динамическая установка раздела таблицы:

public class DayPartitionFunc implements SerializableFunction<BoundedWindow, String> {

String destination = "";

public DayPartitionFunc(String dataset, String table) {
    this.destination = dataset + "." + table+ "$";
}

@Override
public String apply(BoundedWindow boundedWindow) {
    // The cast below is safe because CalendarWindows.days(1) produces IntervalWindows.
    String dayString = DateTimeFormat.forPattern("yyyyMMdd")
                                     .withZone(DateTimeZone.UTC)
                                     .print(((IntervalWindow) boundedWindow).start());
    return destination + dayString;
}}

Есть ли лучший способ достичь того же результата?

Я считаю, что можно использовать декоратор разделов, когда вы не используете потоковую передачу. Мы активно работаем над поддержкой декораторов разделов посредством потоковой передачи. Пожалуйста, дайте нам знать, если вы видите какие-либо ошибки сегодня в режиме без потоковой передачи.

Apache Beam версии 2.0 поддерживает сегментирование выходных таблиц BigQuery прямо из коробки.

Если вы передадите имя таблицы в формате table_name_YYYYMMDD, BigQuery будет рассматривать ее как сегментированную таблицу, которая может имитировать функции таблицы разделов. См. Документацию: https://cloud.google.com/bigquery/docs/partitioned-tables

Я записал данные в секционированные таблицы bigquery через поток данных. Эти записи являются динамическими, если данные в этом разделе уже существуют, я могу либо добавить к нему, либо перезаписать его.

Я написал код на Python. Это операция записи в bigquery в пакетном режиме.

client = bigquery.Client(project=projectName)
dataset_ref = client.dataset(datasetName)
table_ref = dataset_ref.table(bqTableName)       
job_config = bigquery.LoadJobConfig()
job_config.skip_leading_rows = skipLeadingRows
job_config.source_format = bigquery.SourceFormat.CSV
if tableExists(client, table_ref):            
    job_config.autodetect = autoDetect
    previous_rows = client.get_table(table_ref).num_rows
    #assert previous_rows > 0
    if allowJaggedRows is True:
        job_config.allowJaggedRows = True
    if allowFieldAddition is True:
        job_config._properties['load']['schemaUpdateOptions'] = ['ALLOW_FIELD_ADDITION']
    if isPartitioned is True:
        job_config._properties['load']['timePartitioning'] = {"type": "DAY"}
    if schemaList is not None:
        job_config.schema = schemaList            
    job_config.write_disposition = bigquery.WriteDisposition.WRITE_TRUNCATE
else:            
    job_config.autodetect = autoDetect
    job_config._properties['createDisposition'] = 'CREATE_IF_NEEDED'
    job_config.schema = schemaList
    if isPartitioned is True:             
        job_config._properties['load']['timePartitioning'] = {"type": "DAY"}
    if schemaList is not None:
        table = bigquery.Table(table_ref, schema=schemaList)            
load_job = client.load_table_from_uri(gcsFileName, table_ref, job_config=job_config)        
assert load_job.job_type == 'load'
load_job.result()       
assert load_job.state == 'DONE'

Работает нормально.