Create a streaming pipeline using a Dataflow template

Create a BigQuery dataset and table

Create a BigQuery dataset and table with the appropriate schema for your Pub/Sub topic using the Google Cloud console.

In this example, the name of the dataset is taxirides and the name of the table is realtime. To create this dataset and table, follow these steps:

1. Go to the BigQuery page.
   Go to BigQuery
2. In the Explorer panel, next to the project where you want to create the dataset, click more_vertView actions, and then click Create dataset.
3. On the Create dataset panel, follow these steps:
1. For Dataset ID, enter taxirides. Dataset IDs are unique for each Google Cloud project.
2. For Location type, choose Multi-region, and then select US (multiple regions in United States). Public datasets are stored in the US multi-region location. For simplicity, place your dataset in the same location.
3. Leave the other default settings, and then click Create dataset
4. In the Explorer panel, expand your project.
5. Next to your taxirides dataset, click more_vertView actions, and then click Create table.
6. On the Create table panel, follow these steps:
1. In the Source section, for Create table from, select Empty table.
2. In the Destination section, for Table, enter realtime.
3. In the Schema section, click the Edit as text toggle and paste the following schema definition into the box:
   

   ride_id:string,point_idx:integer,latitude:float,longitude:float,timestamp:timestamp, meter_reading:float,meter_increment:float,ride_status:string,passenger_count:integer

4. In the Partition and cluster settings section, for Partitioning, select the timestamp field.
7. Leave the other default settings in place and click Create table.

Run the pipeline

Run a streaming pipeline using the Google-provided Pub/Sub to BigQuery template. The pipeline gets incoming data from the input topic.

1. Go to the Dataflow Jobs page.
   Go to Jobs
2. Click Create job from template.
3. Enter taxi-data as the Job name for your Dataflow job.
4. For Dataflow template, select the Pub/Sub to BigQuery template.
5. For BigQuery output table, enter the following:

   PROJECT_ID:taxirides.realtime

   Replace PROJECT_ID with the project ID of the project where you created your BigQuery dataset.

6. In the section Optional source parameters, for Input Pub/Sub topic, click Enter topic manually.
7. In the dialog, for Topic name enter the following, and then click Save:

   projects/pubsub-public-data/topics/taxirides-realtime

   This publicly available Pub/Sub topic is based on the NYC Taxi & Limousine Commission's open dataset. The following is a sample message from this topic, in the JSON format:

   {"ride_id":"19c41fc4-e362-4be5-9d06-435a7dc9ba8e", "point_idx":217, "latitude":40.75399, "longitude":-73.96302, "timestamp":"2021-03-08T02:29:09.66644-05:00", "meter_reading":6.293821, "meter_increment":0.029003782, "ride_status":"enroute", "passenger_count":1}

8. For Temp location, enter the following:

   gs://BUCKET_NAME/temp/

   Replace BUCKET_NAME with the name of your Cloud Storage bucket. The temp folder stores temporary files, like the staged pipeline job.

9. If your project does not have a default network, enter a Network and a Subnetwork. For more information, see Specify a network and subnetwork.
10. Click Run job.

View your results

1. Go to the BigQuery page.

   Go to BigQuery

2. Click add_boxCompose a new query. A new Editor tab opens.

   SELECT*FROM`PROJECT_ID.taxirides.realtime` WHERE`timestamp`>TIMESTAMP_SUB(CURRENT_TIMESTAMP(),INTERVAL1DAY) LIMIT1000

   Replace PROJECT_ID with the project ID of the project where you created your BigQuery dataset. It can take up to five minutes for data to start appearing in your table.

3. Click Run.

   The query returns rows that have been added to your table in the past 24 hours. You can also run queries using standard SQL.