Configuring a Pipeline

Data CollectorControl Hub uses the alphanumeric characters entered for the pipeline title as a prefix for the generated pipeline ID. For example, if you enter My Pipeline *&%&^^ 123 as the pipeline title, then the pipeline ID has the following value: MyPipeline123tad9f592-5f02-4695-bb10-127b2e41561c.

You can edit the pipeline title. However, because the pipeline ID is used to identify the pipeline, any changes to the pipeline title are not reflected in the pipeline ID.

Use labels to group similar pipelines. For example, you might want to group pipelines by database schema or by the test or production environment.

Assign the label templates to pipelines that you want to use as user-defined sample pipelines.

<label1>/<label2>/<label3>

• Standalone - A single Data Collector process runs the pipeline.
• Cluster Batch - Data Collector spawns additional workers as needed to process data in HDFS or MapR. Processes all available data and then stops the pipeline.
• Cluster Yarn Streaming - Data Collector spawns additional workers as needed to process data, by default. You can limit the number of workers with the Worker Count cluster property. And you can use the Extra Spark Configuration property to pass Spark configurations to the spark-submit script.

  Use to stream data from a Kafka or MapR cluster that uses Spark Streaming on YARN.

• Cluster Mesos Streaming - Data Collector spawns additional workers as needed to process data. Use to stream data from a Kafka cluster that uses Spark Streaming on Mesos.
• Edge - A single Data Collector Edge (SDC Edge) process runs the pipeline on an edge device.

For edge pipelines, the URL to the SDC Edge installation. Used when you directly publish and manage the pipeline on an SDC Edge that is not registered with Control Hub. You can leave the default value when you publish the pipeline to Control Hub, add the pipeline to a job, and start the job on a registered SDC Edge. For more information, see Publishing a Single Pipeline and Manage Pipelines on SDC Edge.

Used only for edge pipelines designed in Data Collector and then published directly to and managed on an SDC Edge that is not registered with Control Hub. You can leave the default value for edge pipelines designed in Control Hub.

Not available in Data Collector pipelines.

• At Least Once - Ensures all data is processed and written to the destination. Might result in duplicate rows.
• At Most Once - Ensures that data is not reprocessed to prevent writing duplicate data to the destination. Might result in missing rows.

Default is At Least Once.

To enable the use of a test origin, select the origin to access the test data, then configure the origin properties on the Test Origin tab. You can use any available origin.

Default is the Dev Raw Data Source origin.

• Discard - Use when you don't want to use the event.
• An executor - To use the event to trigger a task, select the executor that you want to use. For more information about the executors, see Executors.
• Write to Another Pipeline - Use to pass the event to another pipeline for more complex processing.

Use in standalone Data Collector pipelines only.

For more information about pipeline events, see Pipeline Event Generation.

Not available in Data Collector Edge pipelines.

• Discard - Use when you don't want to use the event.
• An executor - To use the event to trigger a task, select the executor that you want to use. For more information about the executors, see Executors.
• Write to Another Pipeline - Use to pass the event to another pipeline for more complex processing.

Use in standalone Data Collector pipelines only.

For more information about pipeline events, see Pipeline Event Generation.

Not available in Data Collector Edge pipelines.

Not available in Data Collector Edge pipelines.

The wait time between retries starts at 15 seconds and doubles until reaching five minutes.

Not available in Data Collector Edge pipelines.

Default is 0.

Not available in Data Collector Edge pipelines.

Use 0 for no limit. When set to 0, Data Collector generates up to the maximum number of threads or concurrency configured in the origin.

You can use this property to help tune pipeline performance. For more information, see Tuning Threads and Runners.

Default is 0.

Not available in Data Collector Edge pipelines.

Minimum number of seconds a pipeline runner waits when idle before generating an empty batch. The number of empty batches that are generated by pipeline runners displays as the Idle Batch Count in the monitor mode runtime statistics.

Use to ensure that batches are generated periodically, even when no data needs to be processed.

Use -1 to allow pipeline runners to wait indefinitely when idle without generating empty batches.

For standalone pipelines only.

Not available in Data Collector Edge pipelines.

Not available in Data Collector Edge pipelines.

• GET
• PUT
• POST
• DELETE
• HEAD

Use any valid content type.

You can use webhook parameters in the payload to include information about the triggering event, such as the pipeline name or state. Enclose webhook parameters in double curly brackets as follows: {{PIPELINE_STATE}}.

Use Basic for Form authentication.

Use one of the following options:

• Discard - Discards error records.
• Send Response to Origin - Passes error records back to the microservice origin to be included in a response to the originating REST API client. Use in microservice pipelines only.
• Write to Amazon S3 - Writes error records to Amazon S3.
• Write to Another Pipeline - Writes error records to another pipeline. To use this option, you need an SDC RPC destination pipeline to process the error records.
• Write to Azure Event Hub - Writes error records to the specified Microsoft Azure Event Hub.
• Write to Elasticsearch - Writes error records to the specified Elasticsearch cluster.
• Write to File - Writes error records to a file in the specified directory.

  Write to File is not supported for cluster mode at this time.

• Write to Google Cloud Storage - Writes error records to Google Cloud Storage.
• Write to Google Pub/Sub - Writes error records to Google Pub/Sub.
• Write to Kafka - Writes error records to the specified Kafka cluster.
• Write to Kinesis - Writes error records to the specified Kinesis stream.
• Write to MapR Streams - Writes error records to the specified MapR Streams cluster.
• Write to MQTT - Writes error records to the specified MQTT broker.

In Data Collector Edge pipelines, you can use Discard, Write to File, or Write to MQTT.

To connect to an external system, you can select a connection that contains the details, or you can directly enter the details in the pipeline. When you select a connection, Control Hub hides other properties so that you cannot directly enter connection details in the pipeline.

To create a new connection, click the Add New Connection icon: . To view and edit the details of the selected connection, click the Edit Connection icon: .

• AWS Keys - Authenticates using an AWS access key pair.
• Instance Profile - Authenticates using an instance profile associated with the Data Collector EC2 instance.
• None - Connects to a public bucket using no authentication.

Amazon resource name (ARN) of the role to assume, entered in the following format:

arn:aws:iam::<account_id>:role/<role_name>

Where <account_id> is the ID of your AWS account and <role_name> is the name of the role to assume. You must create and attach an IAM trust policy to this role that allows the role to be assumed.

Available when assuming another role.

Optional name for the session created by assuming a role. Overrides the default unique identifier for the session.

Available when assuming another role.

Maximum number of seconds for each session created by assuming a role. The session is refreshed if the pipeline continues to run for longer than this amount of time.

Set to a value between 3,600 seconds and 43,200 seconds.

Available when assuming another role.

Sets a session tag to record the name of the currently logged in Data Collector or Transformer user that starts the pipeline or the Control Hub user that starts the job for the pipeline. AWS IAM verifies that the user account set in the session tag can assume the specified role.

Select only when the IAM trust policy attached to the role to be assumed uses session tags and restricts the session tag values to specific user accounts.

When cleared, the connection does not set a session tag.

Available when assuming another role.

When cleared, the stage uses the Amazon S3 default global endpoint, s3.amazonaws.com.

Enter a bucket name or define an expression that evaluates to bucket names.

When using datetime variables in the expression, be sure to configure the time basis for the stage.

Use a specific partition prefix or define an expression that evaluates to a partition prefix.

When using datetime variables in the expression, be sure to configure the time basis for the stage.

You can include periods within the suffix, but do not start the suffix with a period. Forward slashes are not allowed.

Not available for the whole file data format.

• SSE-S3 - Use Amazon S3-managed keys.
• SSE-KMS - Use Amazon Web Services KMS-managed keys.
• SSE-C - Use customer-provided keys.

Default is SSE-S3.

<arn>:<aws>:<kms>:<region>:<acct ID>:<key>/<key ID>

Used for SSE-KMS encryption only.

Default is slash ( / ).

Used for SSE-KMS encryption only.

Used for SSE-C encryption only.

Used for SSE-C encryption only.

Time zone for the destination system. Used with the time basis to resolve datetimes in a time-based bucket or partition prefix.

When the Bucket and Partition Prefix properties have no time component, you can ignore this property.

Default is ${time:now()}.

Default is 10 seconds.

When writing to multiple partitions, setting this property up to the number of partitions being written to can improve performance.

For more information about this and the following properties, see the Amazon S3 TransferManager documentation.

Use a single RPC connection for each destination pipeline. Using simple or bulk edit mode, add additional connections as needed.

Use the port number when you configure the SDC RPC origin that receives the data.

Using simple or bulk edit mode, click the Add icon to add additional certificates.

Path to the local truststore file. Enter an absolute path to the file or enter the following expression to define the file stored in the Data Collector resources directory:

${runtime:resourcesDirPath()}/truststore.jks

By default, no truststore is used.

• Java Keystore File (JKS)
• PKCS #12 (p12 file)

Default is Java Keystore File (JKS).

Default is SunX509.

Enter the Java Secure Socket Extension (JSSE) name for the additional cipher suites that you want to use.

When the destination cannot write the batch within the configured number of retries, it fails the batch.

Default is 3.

The value that you enter increases exponentially after each retry, until it reaches the maximum wait time of 5 minutes. For example, if you set the back off period to 10, the destination attempts the first retry after waiting 10 milliseconds, attempts the second retry after waiting 100 milliseconds, and attempts the third retry after waiting 1,000 milliseconds.

Set to 0 to retry immediately.

Default is 0.

The destination retries the connection based on the Retries Per Batch property.

Default is 5000 milliseconds.

The destination retries the write based on the Retries Per Batch property.

Default is 2000 milliseconds.

To retrieve the policy name, when logged into the Azure portal, navigate to your namespace and event hub, and then click Shared Access Policies for a list of policies.

When appropriate, you can use the default shared access key policy, RootManageSharedAccessKey.

To retrieve a connection string key, after accessing the list of shared access policies, click the policy name, and then copy the Connection String - Primary Key value.

The value typically begins with "Endpoint".

To connect to an external system, you can select a connection that contains the details, or you can directly enter the details in the pipeline. When you select a connection, Control Hub hides other properties so that you cannot directly enter connection details in the pipeline.

To create a new connection, click the Add New Connection icon: . To view and edit the details of the selected connection, click the Edit Connection icon: .

http://<host1>,http://<host2>

You can specify a port number in the URLs to override the default port defined in the HTTP Port property, as follows:

http://<host1>:<port>,http://<host2>:<port>

When a port number is defined in both this property and in the HTTP Port property, the port in this property takes precedence. For example, if you define this property as follows:

http://server1,http://server2:1234

And you define the default HTTP Port property as 9200, then server1 uses the default port of 9200 and server2 uses the port 1234.

The default HTTP port is 9200. The default HTTPS port is 443.

For example, if you enter customer as the index, the destination writes the document within the customer index.

If you use datetime variables in the expression, make sure to configure the time basis appropriately. For details about datetime variables, see Datetime Variables.

For example, if you enter user as the mapping type, the destination writes the document with a user mapping type.

By default, the destination allows Elasticsearch to create the ID.

Selecting this property is the equivalent to setting the client.transport.sniff Elasticsearch property to true.

Use only when Data Collector shares the same network as the Elasticsearch cluster. Do not use for Elastic Cloud or Docker clusters.

• ${time:now()} - Uses the processing time as the time basis. The processing time is the time associated with the Data Collector running the pipeline.
• An expression that calls a field and resolves to a datetime value, such as ${record:value(<date field path>)}. Uses the datetime result as the time basis.

When the Index property does not include datetime variables, you can ignore this property.

Default is ${time:now()}.

Use to establish a parent-child relationship between documents in the same index.

Elasticsearch routes a document to a particular shard in an index based on the routing value defined for the document. You can define a custom value for each document. If you don’t define a custom routing value, Elasticsearch uses the parent ID (if defined) or the document ID as the routing value.

Character encoding of the data to be processed.

For example, you can use the _retry_on_conflict field to specify how many times an update is retried when there is a version conflict. To specify three retries, include the following:

"_retry_on_conflict" : 3

You can use record functions and delimited data record functions when configuring this property.

For more information, see the Elasticsearch documentation.

• Basic - Authenticate with Elasticsearch user name and password. Select this option for Elasticsearch clusters outside of Amazon Elasticsearch Service.
• AWS Signature V4 - Authenticate with AWS. Select this option for Elasticsearch clusters within Amazon Elasticsearch Service.

Available when using Basic authentication.

Available when using Basic authentication.

Available when using AWS Signature V4 authentication.

Available when using AWS Signature V4 authentication.

Available when using AWS Signature V4 authentication.

Configuring this property is the equivalent to configuring the shield.ssl.truststore.path Elasticsearch property.

Not necessary for Elastic Cloud clusters.

Configuring this property is the equivalent to configuring the shield.ssl.truststore.password Elasticsearch property.

Not necessary for Elastic Cloud clusters.

Uses the prefix sdc-${sdc:id()} by default. The prefix evaluates to sdc-<Data Collector ID>. This provides default differentiation in case several Data Collectors write to the same directory.

The Data Collector ID is stored in the following file: $SDC_DATA/sdc.id file.

For more information about environment variables, see Data Collector EnvironmentJava and Security ConfigurationData Collector Environment Configuration in the Data Collector documentation.

You can enter a number of seconds or use the default expression to enter the time in minutes.

Use 0 to avoid using this property.

Use a specific partition prefix or define an expression that evaluates to a partition prefix.

When using datetime variables in the expression, be sure to configure the time basis for the stage.

Time zone for the destination system. Used to resolve datetimes in a time-based partition prefix.

When the Partition Prefix property has no time component, you can ignore this property.

Default is ${time:now()}.

Not required for the whole file data format.

To connect to an external system, you can select a connection that contains the details, or you can directly enter the details in the pipeline. When you select a connection, Control Hub hides other properties so that you cannot directly enter connection details in the pipeline.

To create a new connection, click the Add New Connection icon: . To view and edit the details of the selected connection, click the Edit Connection icon: .

• Default credentials provider - Uses Google Cloud default credentials.
• Service account credentials file (JSON) - Uses credentials stored in a JSON service account credentials file.
• Service account credentials (JSON) - Uses JSON-formatted credentials information from a service account credentials file.

Enter a path relative to the Data Collector resources directory, $SDC_RESOURCES, or enter an absolute path.

Enter JSON-formatted credential information in plain text, or use an expression to call the information from runtime resources or a credential store.credential store. For more information about credential stores, see Credential Stores in the Data Collector documentation.

To connect to an external system, you can select a connection that contains the details, or you can directly enter the details in the pipeline. When you select a connection, Control Hub hides other properties so that you cannot directly enter connection details in the pipeline.

To create a new connection, click the Add New Connection icon: . To view and edit the details of the selected connection, click the Edit Connection icon: .

• Default credentials provider - Uses Google Cloud default credentials.
• Service account credentials file (JSON) - Uses credentials stored in a JSON service account credentials file.
• Service account credentials (JSON) - Uses JSON-formatted credentials information from a service account credentials file.

Enter a path relative to the Data Collector resources directory, $SDC_RESOURCES, or enter an absolute path.

Enter JSON-formatted credential information in plain text, or use an expression to call the information from runtime resources or a credential store.credential store. For more information about credential stores, see Credential Stores in the Data Collector documentation.

Default is 1000.

Default is 100.

Default is 1.

Set to 0 to never control the flow based on message count. To control the message flow when using batch processing, set to a number larger than the message count threshold.

Set to 0 to never control the flow based on message size. To control the message flow when using batch processing, set to a number larger than the request bytes threshold.

• Throw Exception - Triggers pipeline error processing.
• Block - Stops processing new messages until stored messages have been successfully written.
• Ignore - Discards new messages until stored messages have been successfully written.

To connect to an external system, you can select a connection that contains the details, or you can directly enter the details in the pipeline. When you select a connection, Control Hub hides other properties so that you cannot directly enter connection details in the pipeline.

To create a new connection, click the Add New Connection icon: . To view and edit the details of the selected connection, click the Edit Connection icon: .

To ensure a pipeline can connect to Kafka in case a specified broker goes down, list as many brokers as possible.

Not available when using runtime topic resolution.

Use an asterisk (*) to allow writing to any topic name. By default, all topic names are valid.

• Round Robin - Takes turns writing to different partitions.
• Random - Writes to partitions randomly.
• Expression - Uses an expression to write data to different partitions. Writes records to the partitions specified by the results of the expression.
• Default - Uses an expression to extract a partition key from the record. Writes records to partitions based on a hash of the partition key.

When using the default partition strategy, specify an expression that returns the partition key from the record. The expression must evaluate to a string value.

When using the expression partition strategy, specify an expression that evaluates to the partition where you want each record written. Partition numbers start with 0. The expression must evaluate to a numeric value.

Optionally, click Ctrl + Space Bar for help with creating the expression.

Use the property names and values as expected by Kafka. Do not use the broker.list property.

Enter an expression that specifies the attribute where the message keys are stored.

${record:attribute('<message key attribute name>'}

${avro:decode(record:attribute('avroKeySchema'),base64:decodeBytes(record:attribute('<messsage key attribute name')))}

For more information, about working with Kafka message keys, see Kafka Message Keys.

• None (Security Protocol=PLAINTEXT) - Uses no authentication or encryption.
• SSL/TLS Encryption (Security Protocol=SSL)
• SSL/TLS Encryption and Authentication (Security Protocol=SSL)
• Kerberos Authentication (Security Protocol=SASL_PLAINTEXT)
• Kerberos Authentication on SSL/TLS (Security Protocol=SASL_SSL)

Enabling security requires completing several prerequisite tasks and configuring additional security properties, as described in Security in Kafka Stages.

For more information about working with Kafka message key values, see Kafka Message Keys.

To connect to an external system, you can select a connection that contains the details, or you can directly enter the details in the pipeline. When you select a connection, Control Hub hides other properties so that you cannot directly enter connection details in the pipeline.

To create a new connection, click the Add New Connection icon: . To view and edit the details of the selected connection, click the Edit Connection icon: .

• AWS Keys - Authenticates using an AWS access key pair.
• Instance Profile - Authenticates using an instance profile associated with the Data Collector EC2 instance.

Amazon resource name (ARN) of the role to assume, entered in the following format:

arn:aws:iam::<account_id>:role/<role_name>

Where <account_id> is the ID of your AWS account and <role_name> is the name of the role to assume. You must create and attach an IAM trust policy to this role that allows the role to be assumed.

Available when assuming another role.

Optional name for the session created by assuming a role. Overrides the default unique identifier for the session.

Available when assuming another role.

Maximum number of seconds for each session created by assuming a role. The session is refreshed if the pipeline continues to run for longer than this amount of time.

Set to a value between 3,600 seconds and 43,200 seconds.

Available when assuming another role.

Sets a session tag to record the name of the currently logged in Data Collector or Transformer user that starts the pipeline or the Control Hub user that starts the job for the pipeline. AWS IAM verifies that the user account set in the session tag can assume the specified role.

Select only when the IAM trust policy attached to the role to be assumed uses session tags and restricts the session tag values to specific user accounts.

When cleared, the connection does not set a session tag.

Available when assuming another role.

When you add a configuration property, enter the exact property name and the value. The stage does not validate the property names or values.

• Random - Generates a random partition key.

• Expression - Uses the result of an expression as the partition key.

Use for the expression partition strategy.

Not available when using runtime topic resolution.

Use an asterisk (*) to allow writing to any topic name. By default, all topic names are valid.

• Round Robin - Takes turns writing to different partitions.
• Random - Writes to partitions randomly.
• Expression - Uses an expression to write data to different partitions. Writes records to the partitions specified by the results of the expression.
• Default - Uses an expression to extract a partition key from the record. Writes records to partitions based on a hash of the partition key.

When using the default partition strategy, specify an expression that returns the partition key from the record. The expression must evaluate to a string value.

When using the expression partition strategy, specify an expression that evaluates to the partition where you want each record written. Partition numbers start with 0. The expression must evaluate to a numeric value.

Optionally, click Ctrl + Space Bar for help with creating the expression.

Use the property names and values as expected by MapR.

You can use MapR Streams properties and the set of Kafka properties supported by MapR Streams.

<tcp | ssl>://<hostname>:<port>

Use ssl for secure connections to the broker.

tcp://localhost:1883

${pipeline:id()}

If a pipeline includes multiple MQTT stages and you want to use the unique pipeline ID as the client ID for both stages, prefix the client ID with a string like this:

sub-${pipeline:id()} and pub-${pipeline:id()} 

• At Most Once (0)
• At Least Once (1)
• Exactly Once (2)

For more information, see the HiveMQ documentation on quality of service levels.

• Memory - Store messages in memory on the Data Collector machine until the delivery of the message is complete.
• File - Store messages in a local file on the Data Collector machine until the delivery of the message is complete.

Not used when the quality of service level is at most once.

For more information, see the HiveMQ documentation on client persistence.

The user who starts Data Collector must have read and write access to this directory.

Default is 60 seconds.

When selected, the MQTT broker retains the last message published by the destination. Any messages published earlier are lost. When cleared, all messages published by the destination are lost.

For more information about MQTT retained messages, see http://www.hivemq.com/blog/mqtt-essentials-part-8-retained-messages.

Using simple or bulk edit mode, click the Add icon to add additional certificates.

Path to the local truststore file. Enter an absolute path to the file or enter the following expression to define the file stored in the Data Collector resources directory:

${runtime:resourcesDirPath()}/truststore.jks

By default, no truststore is used.

• Java Keystore File (JKS)
• PKCS #12 (p12 file)

Default is Java Keystore File (JKS).

Default is SunX509.

Enter the Java Secure Socket Extension (JSSE) name for the additional cipher suites that you want to use.