Source coordination

Source coordination is the concept of coordinating and distributing work between Data Prepper data sources in a multi-node environment. Some data sources, such as Amazon Kinesis or Amazon Simple Queue Service (Amazon SQS), handle coordination natively. Other data sources, such as OpenSearch, Amazon Simple Storage Service (Amazon S3), Amazon DynamoDB, and JDBC/ODBC, do not support source coordination.

Data Prepper source coordination decides which partition of work is performed by each node in the Data Prepper cluster and prevents duplicate partitions of work.

Inspired by the Kinesis Client Library, Data Prepper utilizes a distributed store in the form of a lease to handle the distribution and deduplication of work.

Formatting partitions

Source coordination separates sources into “partitions of work.” For example, an S3 object would be a partition of work for Amazon S3, or an OpenSearch index would be a partition of work for OpenSearch.

Data Prepper takes each partition of work that is chosen by the source and creates corresponding items in the distributed store that Data Prepper uses for source coordination. Each of these items has the following standard format, which can be extended by the distributed store implementation.

Acquiring partitions

Partitions are acquired in the order that they are returned in the List<PartitionIdentifer> provided by the source. When a node attempts to acquire a partition, Data Prepper performs the following steps:

1. Data Prepper queries the ASSIGNED partitions to check whether any ASSIGNED partitions have expired partition owners. This is intended to assign priority to partitions that have had nodes crash in the middle of processing, which can allow for using a partition state that may be time sensitive.
2. After querying ASSIGNED partitions, Data Prepper queries the CLOSED partitions to determine whether any of the partition’s reOpenAt timestamps have been reached.
3. If there are no ASSIGNED or CLOSED partitions available, then Data Prepper queries the UNASSIGNED partitions until on of these partitions is ASSIGNED.

If this flow occurs and no partition is acquired by the node, then the partition supplier function provided in the getNextPartition method of SourceCoordinator will create new partitions. After the supplier function completes, Data Prepper again queries the partitions for ASSIGNED, CLOSED, and UNASSIGNED.

Global state

Any function that is passed to the getNextPartition method creates new partitions with a global state of Map<String, Object>. This state is shared between all of the nodes in the cluster and will only be run by a single node at a time, as determined by the source.

Configuration

The following table provide optional configuration values for source_coordination.

Supported stores

As of Data Prepper 2.4, only in_memory and dynamodb stores are supported:

• The in_memory store is the default when no source_coordination settings are configured in the data-prepper-config.yaml file and should only be used for single-node configurations.
• The dynamodb store is used for multi-node Data Prepper environments. The dynamodb store can be shared between one or more Data Prepper clusters that need to utilize source coordination.

DynamoDB store

Data Prepper will attempt to create the dynamodb table on startup unless the skip_table_creation flag is configured to true. Optionally, you can configure the time-to-live (ttl) on the table, which results in the store cleaning up items over time. Some sources rely on source coordination for the deduplication of data, so be sure to configure a large enough ttl for the pipeline duration.

If ttl is not configured on the table, any items no longer needed in the table must be cleaned manually.

The following shows the full set of permissions needed for Data Prepper to create the table, enable ttl, and interact with the table:

{
  "Sid": "ReadWriteSourceCoordinationDynamoStore",
  "Effect": "Allow",
  "Action": [
    "dynamodb:DescribeTimeToLive",
    "dynamodb:UpdateTimeToLive",
    "dynamodb:DescribeTable",
    "dynamodb:CreateTable",
    "dynamodb:GetItem",
    "dynamodb:PutItem",
    "dynamodb:Query"
  ],
  "Resource": [
    "arn:aws:dynamodb:${REGION}:${AWS_ACCOUNT_ID}:table/${TABLE_NAME}",
    "arn:aws:dynamodb:${REGION}:${AWS_ACCOUNT_ID}:table/${TABLE_NAME}/index/source-status"
  ]
}

The following example shows a dynamodb store:

source_coordination:
  store:
     dynamodb:
       table_name: "DataPrepperSourceCoordinationStore"
       region: "us-east-1"
       sts_role_arn: "arn:aws:iam::##########:role/SourceCoordinationTableRole"
       ttl: "P7D"
       skip_table_creation: true

In-memory store (default)

The following example shows an in_memory store, which is best used with a single-node cluster:

source_coordination:
  store:
    in_memory:

Metrics

Source coordination metrics are interpreted differently depending on which source is configured. The format of a source coordination metric is <sub-pipeline-name>_source_coordinator_<metric-name>. You can use the sub-pipeline name to identify the source for these metrics because each sub-pipeline is unique to each source.

Progress metrics

The following are metrics related to partition progress:

• partitionsCreatedCount: The number of partition items that have been created. For an S3 scan, this is the number of objects that have had partitions created for them.
• partitionsCompleted: The number of partitions that have been fully processed and marked as COMPLETED. For an S3 scan, this is the number of objects that have been processed.
• noPartitionsAcquired: The number of times that a node has attempted to acquire a partition on which to perform work but has found no available partitions in the store. Use this to indicate that there is no more data coming into the source.
• partitionsAcquired: The number of partitions that have been acquired by nodes on which to perform work. In non-error scenarios, this should be equal to the number of partitions created.
• partitionsClosed: The number of partitions that have been marked as CLOSED. This is only applicable to sources that use the CLOSED functionality.

The following are metrics related to partition errors:

• partitionNotFoundErrors: Indicates that a partition item that is actively owned by a node does not have a corresponding store item. This should only occur if an item in the table has been manually deleted.
• partitionNotOwnedErrors: Indicates that a node that owns a partition has lost ownership due to the partition ownership timeout expiring. Unless the source is able to checkpoint the partition with saveState, this error results in duplicate item processing.
• partitionUpdateErrors: The number of errors that were received when an update to the store for this partition item failed. Is prefixed with either saveState, close, or complete to indicate which update action is failing.