google bigquery - Escribir diferentes valores en diferentes tablas de BigQuery en Apache Beam

google-bigquery google-cloud-dataflow

Question

Supongamos que tengo una PCollection<Foo> y quiero escribirla en varias tablas BigQuery, eligiendo una tabla potencialmente diferente para cada Foo .

¿Cómo puedo hacer esto usando la API Apache Beam BigQueryIO ?

Answer 1

A partir de Beam 2.12.0, esta característica también está disponible en Python SDK. Está marcado como experimental, por lo que deberá pasar --experiments use_beam_bq_sink para habilitarlo. Harías algo así:

def get_table_name(element):
  if meets_some_condition(element):
    return ''mytablename1''
  else:
    return ''mytablename2''


p = beam.Pipeline(...)

my_input_pcoll = p | ReadInMyPCollection()

my_input_pcoll | beam.io.gcp.bigquery.WriteToBigQuery(table=get_table_name)

El nuevo sumidero admite una serie de otras opciones, que puede revisar en el pydoc

Answer 2

Esto es posible utilizando una función recientemente agregada a BigQueryIO en Apache Beam.

PCollection<Foo> foos = ...;
foos.apply(BigQueryIO.write().to(new SerializableFunction<ValueInSingleWindow<Foo>, TableDestination>() {
  @Override
  public TableDestination apply(ValueInSingleWindow<Foo> value) {  
    Foo foo = value.getValue();
    // Also available: value.getWindow(), getTimestamp(), getPane()
    String tableSpec = ...;
    String tableDescription = ...;
    return new TableDestination(tableSpec, tableDescription);
  }
}).withFormatFunction(new SerializableFunction<Foo, TableRow>() {
  @Override
  public TableRow apply(Foo foo) {
    return ...;
  }
}).withSchema(...));

Dependiendo de si la entrada PCollection<Foo> está limitada o no, bajo el capó esto creará múltiples trabajos de importación de BigQuery (uno o más por tabla dependiendo de la cantidad de datos), o usará la API de inserciones de transmisión de BigQuery.

La versión más flexible de la API usa DynamicDestinations , que le permite escribir diferentes valores en diferentes tablas con diferentes esquemas, e incluso le permite usar entradas laterales del resto de la tubería en todos estos cálculos.

Además, BigQueryIO se ha refactorizado en una serie de transformaciones reutilizables que puede combinar para implementar casos de uso más complejos: consulte los archivos en el directorio de origen .

Esta característica se incluirá en la primera versión estable de Apache Beam y en la próxima versión de Dataflow SDK (que se basará en la primera versión estable de Apache Beam). En este momento puede usar esto ejecutando su tubería contra una instantánea de Beam at HEAD desde github.