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

Я хотел воспользоваться новой функциональностью BigQuery для таблиц с временным разделением, но я не уверен, что это возможно в настоящее время в версии 1.6 Dataflow SDK.

Рассматривая API-интерфейс BigQuery JSON, для создания разделенной на день таблицы необходимо передать

"timePartitioning": { "type": "DAY" }

вариант, но интерфейс com.google.cloud.dataflow.sdk.io.BigQueryIO позволяет указывать только TableReference.

Я подумал, что, может быть, я мог бы предварительно создать таблицу и пробраться в декоратор раздела через лямбду BigQueryIO.Write.toTableReference..? У кого-нибудь еще есть успех в создании / записи секционированных таблиц через Dataflow?

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

Источник

6 ответов

Решение

Как говорит Паван, определенно возможно писать в таблицы разделов с помощью 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)
        );

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

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 из коробки.

Источник

Я записал данные в разделенные таблицы 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'

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

Источник

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

Источник
Другие вопросы по тегам