Creazione/scrittura nella tabella BigQuery parititoned tramite Google Cloud Dataflow

Question

Volevo sfruttare la nuova funzionalità BigQuery delle tabelle partizionate nel tempo, ma non sono sicuro che ciò sia attualmente possibile nella versione 1.6 dell'SDK Dataflow.

Guardando il BigQuery JSON API, per creare una tabella partizionata giorno si ha la necessità di passare a un'opzione di

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

, ma l'interfaccia com.google.cloud.dataflow.sdk.io.BigQueryIO solo consente di specificare un TableReference.

Ho pensato che forse potevo pre-creare il tavolo e intrufolarmi in un decoratore di partizioni tramite un lambda BigQueryIO.Write.toTableReference? Qualcun altro sta avendo successo con la creazione/scrittura di tabelle partizionate tramite Dataflow?

Questo sembra un problema simile all'impostazione di table expiration time che non è attualmente disponibile.

Answer 1

Come dice Pavan, è sicuramente possibile scrivere su tabelle di partizione con Dataflow. Stai utilizzando lo DataflowPipelineRunner in modalità streaming o batch?

La soluzione che hai proposto dovrebbe funzionare. In particolare, se si crea una tabella con la configurazione del partizionamento della data, è possibile utilizzare un lambda BigQueryIO.Write.toTableReference per scrivere in una partizione data. Per esempio:

/** 
* 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)); 

Answer 2

Credo che dovrebbe essere possibile utilizzare il decoratore di partizioni quando non si utilizza lo streaming. Stiamo lavorando attivamente per supportare i decoratori di partizioni attraverso lo streaming. Facci sapere se stai riscontrando degli errori oggi con la modalità non streaming.

Answer 3

L'approccio che ho preso (funziona in modalità streaming, troppo):

• Definire una finestra personalizzata per il record in entrata

• Convertire la finestra nella tabella partizione/nome

  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) 
       ); 
  

Impostazione della finestra in base i dati in arrivo, il fine immediato possono essere ignorati, come il valore iniziale viene utilizzato per impostare la partizione:

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; 
} 

Impostazione della tabella delle partizioni in modo dinamico:

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; 
}} 

c'è un modo migliore per raggiungere il stesso risultato?

Answer 4

Apache Beam versione 2.0 supporta lo sharding delle tabelle di output BigQuery out of the box.