bigquery operator

[danielnorberg]

Usage:

Run a query and create table from results

final Task<TableId> task = Task.named("task")
    .ofType(TableId.class)
    .context(BigQueryContext.create(table))
    .context(BigQueryOperator.create())
    .process((stagingTable, bq) -> bq.job(
        JobInfo.of(QueryJobConfiguration.newBuilder("SELECT foo FROM input")
            .setDestinationTable(stagingTable.tableId())
            .build()))
        .success(response -> stagingTable.publish()));

Run a query and get the results directly

final Task<String> task = Task.named("task")
    .ofType(String.class)
    .context(BigQueryOperator.create())
    .process(bq -> bq.query("SELECT foo FROM input")
        .success(result -> "success!"));

Load data from GCS (via hades)

final Task<TableId> task = Task.named("load")
    .ofType(TableId.class)
    .input(() -> HadesTasks.lookup("src-endpoint", "src-partition"))
    .context(BigQueryContext.create(TableId.of("dst-project", "dst-dataset", "dst-table")))
    .operator(BigQueryOperator.create())
    .process((input, table, bq) -> bq
        .job(LoadJobConfiguration
            .newBuilder(table.tableId(), input.uri())
            .setSchema(Schema.of(...))) // configure avro schema?
        .success(job -> table.publish()));

[NatashaL]

LGTM 👍