Google Pub/Sub Subscriber

When the Google Pub/Sub Subscriber origin consumes messages from a Google Pub/Sub subscription, it must pass credentials to Google Pub/Sub. Configure the origin to retrieve the credentials from the Google Application Default Credentials or from a Google Cloud service account credentials file.

The Google Pub/Sub Subscriber origin can perform parallel processing and enables the creation of a multithreaded pipeline. The origin uses multiple concurrent threads based on the Num Pipeline Runners property.

When you start the pipeline, each thread connects to the origin system and creates a batch of data, and passes the batch to an available pipeline runner. A pipeline runner is a sourceless pipeline instance - an instance of the pipeline that includes all of the processors and destinations in the pipeline and performs all pipeline processing after the origin.

Each pipeline runner processes one batch at a time, just like a pipeline that runs on a single thread. When the flow of data slows, the pipeline runners wait idly until they are needed, generating an empty batch at regular intervals. You can configure the Runner Idle Time pipeline property specify the interval or to opt out of empty batch generation.

Multithreaded pipelines preserve the order of records within each batch, just like a single-threaded pipeline. But since batches are processed by different pipeline instances, the order that batches are written to destinations is not ensured.

For example, say you set the Num Pipeline Runners property to 5. When you start the pipeline, the origin creates five threads, and Data Collector creates a matching number of pipeline runners. Upon receiving data, the origin passes a batch to each of the pipeline runners for processing.

Each pipeline runner performs the processing associated with the rest of the pipeline. After a batch is written to pipeline destinations, the pipeline runner becomes available for another batch of data. Each batch is processed and written as quickly as possible, independent from other batches processed by other pipeline runners, so batches may be written differently from the read-order.

At any given moment, the five pipeline runners can each process a batch, so this multithreaded pipeline processes up to five batches at a time. When incoming data slows, the pipeline runners sit idle, available for use as soon as the data flow increases.

For more information about multithreaded pipelines, see Multithreaded Pipeline Overview.

The Google Pub/Sub Subscriber origin includes user-defined message attributes in record header attributes when they are available. When the origin processes Avro data, it includes the Avro schema in an avroSchema record header attribute.

A Google Pub/Sub message contains a payload and optional attributes that describe the payload content. If the Google Pub/Sub Subscriber origin consumes a message with optional attributes, the origin includes the message attributes in record header attributes.

You can use the record:attribute or record:attributeOrDefault functions to access the information in the attributes. For more information about working with record header attributes, see Working with Header Attributes.

The Google Pub/Sub Subscriber origin processes data differently based on the data format. Google Pub/Sub Subscriber can process the following types of data:

Avro

Generates a record for every message. Includes a "precision" and "scale" field attribute for each Decimal field. For more information about field attributes, see Field Attributes.

The origin writes the Avro schema to an avroSchema record header attribute. For more information about record header attributes, see Record Header Attributes.

You can use one of the following methods to specify the location of the Avro schema definition:

• Message/Data Includes Schema - Use the schema in the message.
• In Pipeline Configuration - Use the schema that you provide in the stage configuration.
• Confluent Schema Registry - Retrieve the schema from Confluent Schema Registry. The Confluent Schema Registry is a distributed storage layer for Avro schemas. You can configure the origin to look up the schema in the Confluent Schema Registry by the schema ID embedded in the message or by the schema ID or subject specified in the stage configuration.

Using a schema in the stage configuration or retrieving a schema from the Confluent Schema Registry overrides any schema that might be included in the message and can improve performance.

Binary

Generates a record with a single byte array field at the root of the record.

When the data exceeds the user-defined maximum data size, the origin cannot process the data. Because the record is not created, the origin cannot pass the record to the pipeline to be written as an error record. Instead, the origin generates a stage error.

Delimited

Generates a record for each delimited line. You can use the following delimited format types:

• Default CSV - File that includes comma-separated values. Ignores empty lines in the file.
• RFC4180 CSV - Comma-separated file that strictly follows RFC4180 guidelines.
• MS Excel CSV - Microsoft Excel comma-separated file.
• MySQL CSV - MySQL comma-separated file.
• PostgreSQL CSV - PostgreSQL comma-separated file.
• PostgreSQL Text - PostgreSQL text file.
• Tab-Separated Values - File that includes tab-separated values.
• Custom - File that uses user-defined delimiter, escape, and quote characters.

You can use a list or list-map root field type for delimited data, optionally including the header information when available. For more information about the root field types, see Delimited Data Root Field Type.

When using a header line, you can allow processing records with additional columns. The additional columns are named using a custom prefix and integers in sequential increasing order, such as _extra_1, _extra_2. When you disallow additional columns when using a header line, records that include additional columns are sent to error.

You can also replace a string constant with null values.

When a record exceeds the maximum record length defined for the origin, the origin processes the object based on the error handling configured for the stage.

JSON

Generates a record for each JSON object. You can process JSON files that include multiple JSON objects or a single JSON array.

When an object exceeds the maximum object length defined for the origin, the origin processes the object based on the error handling configured for the stage.

Protobuf

Generates a record for every protobuf message. By default, the origin assumes messages contain multiple protobuf messages.

Protobuf messages must match the specified message type and be described in the descriptor file.

When the data for a record exceeds 1 MB, the origin cannot continue processing data in the message. The origin handles the message based on the stage error handling property and continues reading the next message.

For information about generating the descriptor file, see Protobuf Data Format Prerequisites.

SDC Record

Generates a record for every record. Use to process records generated by a Data Collector pipeline using the SDC Record data format.

For error records, the origin provides the original record as read from the origin in the original pipeline, as well as error information that you can use to correct the record.

When processing error records, the origin expects the error file names and contents as generated by the original pipeline.

Text

Generates a record for each line of text or for each section of text based on a custom delimiter.

When a line or section exceeds the maximum line length defined for the origin, the origin truncates it. The origin adds a boolean field named Truncated to indicate if the line was truncated.

For more information about processing text with a custom delimiter, see Text Data Format with Custom Delimiters.

XML

Generates records based on a user-defined delimiter element. Use an XML element directly under the root element or define a simplified XPath expression. If you do not define a delimiter element, the origin treats the XML file as a single record.

Generated records include XML attributes and namespace declarations as fields in the record by default. You can configure the stage to include them in the record as field attributes.

You can include XPath information for each parsed XML element and XML attribute in field attributes. This also places each namespace in an xmlns record header attribute.

Note: Field attributes and record header attributes are written to destination systems automatically only when you use the SDC RPC data format in destinations. For more information about working with field attributes and record header attributes, and how to include them in records, see Field Attributes and Record Header Attributes.

When a record exceeds the user-defined maximum record length, the origin skips the record and continues processing with the next record. It sends the skipped record to the pipeline for error handling.

Use the XML data format to process valid XML documents. For more information about XML processing, see Reading and Processing XML Data.

Tip: If you want to process invalid XML documents, you can try using the text data format with custom delimiters. For more information, see Processing XML Data with Custom Delimiters.