To demonstrate this, we will use a sample dataset of sales transactions, including the state where the transaction took place, and an enrichment table of sales regions. Each sales region covers several states, and we want to annotate each transaction with its sales region, in real time, for immediate analysis and avoiding future joins with the sales_region table. First, create the PubSub topic and BigQuery table:

PROJECT_ID=<my_project>
                  gcloud pubsub topics create live_transactions
                  bq mk --location=us sales_data
                  # load data into the table
                  bq load --autodetect $PROJECT_ID:sales_data.us_sales_regions ./bq_data.csv
                  # examine the table
                  bq head -n 10 $PROJECT_ID:sales_data.us_sales_regions

Next, you’ll need to add a schema to your PubSub topic. This can be done after you switch the BigQuery UI to “Dataflow Engine” mode as indicated in Figure 9-1 and 9-2:

Figure 8-1. Query settings to enable “Dataflow SQL”

Figure 8-2. Enabling Dataflow Engine

Now you should be able to add “Cloud Dataflow sources” as shown in Figure 9-3.

Figure 8-3. Adding Cloud Dataflow sources

Find and choose your topic as in Figure 9-4.

Figure 8-4. Adding your PubSub topic

Select your topic from the resource tree to add a schema as in Figures 9-5 and 9-6.

Figure 8-5. Selecting topic to which attach a schema

Figure 8-6. Adding a schema to your topic

Add the following schema and submit, shown in Figure 10-7. Now you can query on this pubsub topic as if it were a BigQuery table.

Figure 8-7. Adding a schema to your topic

Generate PubSub Data

export PROJECT_ID=$PROJECT_ID
python publish_pubsub.py

Now you can author a fully streaming pipeline in the Dataflow SQL UI and execute the job. This query demonstrates joining a BigQuery table with a PubSub stream, as well as a windowed aggregation with TUMBLE(), the SQL implementation of “sliding windows”.

SELECT
  SUM(stream.amount) AS state_sales,
  stream.state AS state,
  stream.window_start AS minute
FROM
  TUMBLE ( (
    SELECT
      tr.tr_time_str,
      tr.state,
      tr.amount,
      tr.event_timestamp
    FROM
      pubsub.topic.dhodun1.live_transactions AS tr ),
    DESCRIPTOR(event_timestamp),
    "INTERVAL 1 MINUTE" ) AS stream
GROUP BY
  stream.window_start,
  stream.state
 ) stream_aggregate
 INNER JOIN dhodun1.sales_data.us_sales_regions table
 ON stream_aggregate.state = table.state_name
 GROUP BY table.sales_region, stream_aggregate.minute

First, create a Dataproc cluster with the necessary connectors and initialization actions

REGION=us-central1
CLUSTER_NAME=sample-cluster
gcloud dataproc clusters create ${CLUSTER_NAME} 
    --region ${REGION} 
    --initialization-actions gs://goog-dataproc-initialization-actions-${REGION}/connectors/connectors.sh 
    --metadata gcs-connector-version=2.2.0 
    --metadata bigquery-connector-version=1.2.0 
    --metadata spark-bigquery-connector-version=0.19.1

When the cluster is created, prepare a PySpark script to submit that will load an entire table into Spark for additional Querying

#!/usr/bin/python
              from pyspark.sql import SparkSession
              spark = SparkSession 
                .builder 
                .master('yarn') 
                .appName('cookbook-query') 
                .getOrCreate()
              # Use the Cloud Storage bucket for temporary BigQuery export data used
              # by the connector.
              bucket = "dhodun1"
              spark.conf.set('temporaryGcsBucket', bucket)
              # Load data from BigQuery.
              states = spark.read.format('bigquery') 
                .option('table', 'dhodun1:sales_data.us_sales_regions') 
                .load()
              states.createOrReplaceTempView('us_sales_regions')
              # Count how many states in each sales region
              regions_count = spark.sql(
                  'SELECT COUNT(*) AS num_states, sales_region FROM us_sales_regions GROUP BY sales_region')
              regions_count.show()
              regions_count.printSchema()
              # Saving the data to BigQuery
              regions_count.write.format('bigquery') 
                .option('table', 'sales_data.sales_region_aggregate') 
                .save()

Lastly, submit the query using the Dataproc Jobs API. Notice that you are using a pre-built jar and providing the script.

#!/bin/bash
                REGION=us-central1
                CLUSTER_NAME=sample-cluster
                gcloud dataproc jobs submit pyspark query_script.py 
                    --cluster=$CLUSTER_NAME 
                    --region=$REGION 
                    --jars=gs://spark-lib/bigquery/spark-bigquery-latest.jar

You should see job completion, as well as a new table called “sales_region_aggregate” in your BigQuery dataset, as shown in the job output in Figure 9-8.

Figure 8-8. Dataproc job output

First, annotate any POJOs like this so that Beam will detect and register a schema for the object. Note that you can mark fields as nullable:

    @DefaultSchema(JavaFieldSchema.class)
      public class LiveTransaction {
          String tr_time_str;
          @javax.annotation.Nullable String first_name;
          String last_name;
          String city;
          String state;
          String product;
          Double amount;
      }

Now, you can use this schema in your pipeline to avoid using custom coders. You can also convert to and from <Row> objects, which are the generic schematized object in beam:

         pipeline.apply("ReadFromPubSub", PubsubIO.readStrings().fromTopic(topic))
                  .apply("JsonToRow", JsonToRow.withSchema(LiveTransaction))
                  .apply("ConvertToObject", Convert.to(LiveTransaction.class))

Lastly, you can now add the “useBeamSchema()” object on your BigQueryIO.write() calls, whereas previously you had to create a TableSchema object and write out the structure of the data again to then pass to BigQueryIO.write():

.apply("WriteToBQ",
                          BigQueryIO.<LiveTransaction>write().to(output).useBeamSchema()
                                  .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
                                  .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED));

To run the example pipeline, replace the static variables in the pipeline code, start the python generator, and run the pipeline with bash run.sh.

First, branch your pipeline the standard way and implement a Schema-aware filter function to determine which elements you want persisted in which method.

// Streaming Entries
          liveTransactions
                  .apply("FilterLargeAmounts", Filter.<Row>create().whereFieldName("amount", (Double amount) -> amount > amountCutoff))

          // Minibatch Entries
          liveTransactions
                  .apply("FilterSmallAmounts", Filter.<Row>create().whereFieldName("amount", (Double amount) -> amount <= amountCutoff))

Next, implement your BigQuery write as before on the streaming branch

.apply("WriteToBQStreaming",
                          BigQueryIO.<Row>write().to(output).useBeamSchema()
                                  .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
                                  .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED));

Lastly, implement a “BATCH_LOADS” BigQuery writer by setting .withMethod(), .withTriggeringFrequency(), and .withNumFileShards():

.apply("WriteToBQFileLoads",
                          BigQueryIO.<Row>write().to(output).useBeamSchema()
                                  .withMethod(BigQueryIO.Write.Method.FILE_LOADS)
                                  .withTriggeringFrequency(Duration.standardSeconds(minibatchFrequency))
                                  .withNumFileShards(1)
                                  .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
                                  .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED));

Run your pipeline with bash run.sh