Overview
You can use Dataflow Data Pipelines to create recurrent job schedules, understand where resources are spent over multiple job executions, define and manage data freshness objectives, and drill down into individual pipeline stages to fix and optimize your pipelines.
Data Pipeline features:
- Create a recurring batch pipeline to run a batch job on a schedule.
- Create a recurring incremental batch pipeline to run a batch job against the latest version of input data.
- Use the pipeline summary scorecard to view a pipeline's aggregated capacity usage and resource consumption.
- View a streaming pipeline's data freshness. This metric, which evolves over time, can be tied to an alert that notifies you when freshness falls below a specified objective.
- Use pipeline metric graphs to compare batch pipeline jobs and find anomalies.
Data Pipeline use restrictions:
Regional availability: You can create data pipelines in available Cloud Scheduler regions.
Quota:
- Default number of pipelines per project: 500
Default number of pipelines per organization: 2500
The organization level quota is disabled by default. You can opt-in to organization level quotas, and if you do so, each organization can have at most 2500 pipelines by default.
API reference documentation:
For API documentation, see the Data Pipelines reference.
Types of data pipelines
Dataflow has two data pipeline types: streaming and batch. Both types of pipelines run jobs that are defined in Dataflow templates.
- Streaming data pipeline
- A streaming data pipeline runs a Dataflow streaming job immediately after it is created.
- Batch data pipeline
- A batch data pipeline runs a Dataflow batch job on a user-defined schedule. The batch pipeline input filename can be parameterized to allow for incremental batch pipeline processing.
Incremental batch pipelines
You can use datetime placeholders to specify an incremental input file format for a batch pipeline.
- Placeholders for year, month, date, hour, minute, and second can be used, and
must follow the
strftime()
format. Placeholders are preceded by the percentage symbol (%). - Parameter formatting is not verified during pipeline creation.
- Example: If you specify "gs://bucket/Y" as the parameterized input file path,
it will evaluated as "gs://bucket/Y", since "Y" without a preceding "%"
does not map to the
strftime()
format.
- Example: If you specify "gs://bucket/Y" as the parameterized input file path,
it will evaluated as "gs://bucket/Y", since "Y" without a preceding "%"
does not map to the
At each scheduled batch pipeline execution time, the placeholder portion of the input file path is evaluated to the current (or time-shifted) datetime (date values are evaluated using the current date in the time zone of the scheduled job). If the evaluated file path matches the path of an input file, the file will be picked up for processing by the batch pipeline at the scheduled time.
- Example: A batch pipeline is scheduled to repeat at the start of each hour
PST. If you parameterize the input file path as
gs://bucket-name/%Y-%m-%d/prefix-%H_%M.csv
, on April 15, 2021, 6PM PST, the input file path will be evaluated togs://bucket-name/2021-04-15/prefix-18_00.csv
.
Using time shift parameters
You can use + or - minute or hour time shift parameters, enclosed in curly braces with the format, "{[+|-][0-9]+[m|h]}", to support matching an input file path with an evaluated datetime that is shifted before or after the current datetime of the pipeline schedule. The batch pipeline will continue to repeat at its scheduled time, but the input file path will be evaluated with the specified time offset.
- Example: A batch pipeline is scheduled to repeat at the start of each hour
PST. If you parameterize the input file path as
gs://bucket-name/%Y-%m-%d/prefix-%H_%M.csv{-2h}
, on April 15, 2021, 6PM PST, the input file path will be evaluated togs://bucket-name/2021-04-15/prefix-16_00.csv
.
Data pipeline roles
For data pipeline operations to succeed, a user must be granted the necessary IAM roles, as follows:
A user must have the appropriate role to perform operations:
Datapipelines.admin
: Can perform all data pipeline operationsDatapipelines.viewer
: Can view data pipelines and jobsDatapipelines.invoker
: Can invoke a data pipeline job run (this role can be enabled using the API)
A user must be able to act as the service account used by Cloud Scheduler and Dataflow by being granted the
roles/iam.serviceAccountUser
role on that account. If the user does not select a service account for Cloud Scheduler and Dataflow, the default Compute Engine service account is used.
Creating a data pipeline
You can create a data pipeline in two ways:
Data pipelines setup page: When you first access the Dataflow pipelines feature in the Google Cloud console, a setup page opens. Enable the listed APIs to create data pipelines.
Import a job
You can import a Dataflow batch or streaming job that is based on a classic or flex template and make it a data pipeline.
Go to the Dataflow Jobs page in the Google Cloud console, select a completed job, then on the Job Details page, select "+IMPORT AS PIPELINE".
On the Create pipeline from template page, the "data pipeline" pipeline option is selected. Other parameters are populated with the options of the imported job.
- For a batch job, provide a recurrence schedule in the
"Schedule your pipeline" section under Template parameters.
Providing an email account address for the Cloud Scheduler,
which is used to schedule batch runs, is optional. If it is not
specified, the
default Compute Engine service account
is used. Note: the user must be granted the
roles/iam.serviceAccountUser
role on the service account used by Cloud Scheduler, whether it is a user-specified or the default Compute Engine service account (see Data pipeline roles).
- For a batch job, provide a recurrence schedule in the
"Schedule your pipeline" section under Template parameters.
Providing an email account address for the Cloud Scheduler,
which is used to schedule batch runs, is optional. If it is not
specified, the
default Compute Engine service account
is used. Note: the user must be granted the

Create a data pipeline
Go to the Dataflow Pipelines page in the Google Cloud console, then select "+CREATE DATA PIPELINE".
On the Create pipeline from template page under Job management, select "Data pipeline", provide a pipeline name, and fill in the other template selection and parameter fields.
- For a batch job, provide a recurrence schedule in the
"Schedule your pipeline" section under Template parameters.
Providing an email account address for the Cloud Scheduler,
which is used to schedule batch runs, is optional. If it is not
specified, the
default Compute Engine service account
is used. Note: the user must be granted the
roles/iam.serviceAccountUser
role on the service account used by Cloud Scheduler, whether it is a user-specified or the default Compute Engine service account (see Data pipeline roles).
- For a batch job, provide a recurrence schedule in the
"Schedule your pipeline" section under Template parameters.
Providing an email account address for the Cloud Scheduler,
which is used to schedule batch runs, is optional. If it is not
specified, the
default Compute Engine service account
is used. Note: the user must be granted the

Create a batch data pipeline
To create this sample batch data pipeline, you must have access to the following resources in your project:
- A Cloud Storage bucket to store input and output files
- A BigQuery dataset where you will create a table.
This example pipeline uses the Cloud Storage Text to BigQuery batch pipeline template, which reads files in CSV format from Cloud Storage, runs a transform, then inserts values into your-project-id:your-dataset-name.three_column_table.
Create the following files on your local drive:
A
bq_three_column_table.json
file that contains the following schema of the destination BigQuery table.{ "BigQuery Schema": [ { "name": "col1", "type": "STRING" }, { "name": "col2", "type": "STRING" }, { "name": "col3", "type": "INT64" } ] }
A
split_csv_3cols.js
Javascript file, which implements a simple transformation on the input data before insertion into BigQuery.function transform(line) { var values = line.split(','); var obj = new Object(); obj.col1 = values[0]; obj.col2 = values[1]; obj.col3 = values[2]; var jsonString = JSON.stringify(obj); return jsonString; }
A
file01.csv
CSV file with several records that will be inserted into the BigQuery table.b8e5087a,74,27531 7a52c051,4a,25846 672de80f,cd,76981 111b92bf,2e,104653 ff658424,f0,149364 e6c17c75,84,38840 833f5a69,8f,76892 d8c833ff,7d,201386 7d3da7fb,d5,81919 3836d29b,70,181524 ca66e6e5,d7,172076 c8475eb6,03,247282 558294df,f3,155392 737b82a8,c7,235523 82c8f5dc,35,468039 57ab17f9,5e,480350 cbcdaf84,bd,354127 52b55391,eb,423078 825b8863,62,88160 26f16d4f,fd,397783
Use
gsutil
to copy the files to folders in a Cloud Storage bucket in your project, as follows:Copy
bq_three_column_table.json
andsplit_csv_3cols.js
togs://your-bucket/text_to_bigquery/
gsutil cp bq_three_column_table.json gs://<var>your-bucket</var>/text_to_bigquery/
gsutil cp split_csv_3cols.js gs://<var>your-bucket</var>/text_to_bigquery/
Copy
file01.csv
togs://your-bucket/inputs/
gsutil cp file01.csv gs://<var>your-bucket</var>/inputs/
Create a "tmp" folder in your-bucket from the Cloud Storage browser. Select your folder name to open the Bucket details page, then click CREATE FOLDER to create a "tmp" folder in your bucket.
Go to the Dataflow Pipelines page, then select "CREATE DATA PIPELINE". Enter or select the following items on the Create pipeline from template page:
- Job management:
- Select "Data pipeline".
- Pipeline name: Enter "text_to_bq_batch_data_pipeline".
- Click CONTINUE.
- Template selection:
- Regional endpoint: Select a Compute Engine region.
- Template list: Under "Process Data in Bulk (batch)", select "Text File on Cloud Storage to BigQuery. Description: Batch pipeline. Reads text files stored in Cloud Storage, transforms them using a JavaScript user-defined function (UDF), and outputs the result to BigQuery." Note: Don't select the streaming pipeline with the same name under "Process Data Continuously (stream)".
- Click CONTINUE.
- Template parameters:
- Schedule your pipeline: Select a schedule, such as Hourly at Minute 25, in your Timezone. You can edit the schedule after you submit the pipeline, as explained below.
- Required parameters:
- JavaScript UDF path in Cloud Storage:
gs://your-bucket/text_to_bigquery/split_csv_3cols.js
- JSON path:
gs://your-bucket/text_to_bigquery/bq_three_column_table.json
- JavaScript UDF name: "transform"
- BigQuery output table (fully qualified table name):
your_project_id:your_dataset.three_column_table
- Cloud Storage input path:
gs://your_bucket/inputs/file*.csv
- _Temporary BigQuery directory:
gs://your_bucket/tmp
- Temporary location:
gs://your_bucket/tmp
- JavaScript UDF path in Cloud Storage:
- Click SUBMIT.
- Job management:
Confirm pipeline and template information and view current and previous history from the Pipeline details page.
You can also run a batch pipeline on demand using the Run button in the Dataflow Pipelines console.
Create a sample streaming data pipeline
You can create a sample streaming data pipeline by following the sample batch pipeline instructions, with the following differences:
Pipeline schedule. You do not specify a schedule for a streaming data pipeline. The Dataflow streaming job is started immediately.
Template Selection: Under "Process Data Continuously (stream)", select "Text Files on Cloud Storage to BigQuery." Description: A streaming pipeline that can read text files stored in Cloud Storage, perform a transform via a user defined JavaScript function, and stream the results into BigQuery. This pipeline requires a JavaScript function and a JSON representation of the BigQuery TableSchema.
Worker machine type: The pipeline will process the initial set of files matching the
gs://<your_bucket>/inputs/file*.csv
pattern and any additional files matching this pattern that you upload to theinputs/
folder. If the size of CSV files exceeds several GBs, to avoid possible out-of-memory errors, select a machine type with higher memory than the defaultn1-standard-4
machine type, such as then1-highmem-8
.
Investigating pipeline objectives violations
Recurring batch pipelines
On the Pipeline Details page in the Google Cloud console, use the Pipeline status panel's "Individual job status" and "Thread time per step" graphs for an initial analysis of the health of your pipeline.
Sample investigation:
You have a recurring batch pipeline that runs every hour at 3 minutes past the hour, each job normally runs for approximately 9 minutes, and you have an objective for all jobs to complete in less than 10 minutes.
The "job status" graph shows that a job ran for more than 10 minutes.
In the Update/Execution history table, you find the job that ran during the hour of interest, then click through to the Dataflow job details page. On that page you find the longer running stage, and then look in the logs for possible errors to determine the cause of the delay.
Streaming pipelines
Under the PIPELINE INFO TAB on the Pipeline Details page in the Google Cloud console, use the Pipeline status panel's Data Freshness graph for an initial analysis of the health of your pipeline.
Sample investigation:
You have a streaming pipeline that normally produces an output with a data freshness of 20 seconds.
You set an objective of having a 30 second data freshness guarantee. When you review the data freshness graph, you notice that between 9 and 10 AM, data freshness jumped to almost 40 seconds.
You switch to the PIPELINE METRICS tab, then view the Throughput, CPU Utilization, and Memory Utilization graphs for further analysis.