Optional row filter.
When specified, only rows that satisfy the filter condition are read from the delta table. The condition must be a valid SQL Boolean expression that can be used in the `where` clause of the `select * from my_table where ...` query.
| +| `filter` | string | |Optional row filter.
When specified, only rows that satisfy the filter condition are read from the delta table. The condition must be a valid SQL Boolean expression that can be used in the `where` clause of the `select * from my_table where ...` query.
| | `snapshot_filter` | string | |Optional snapshot filter.
This option is only valid when `mode` is set to `snapshot` or `snapshot_and_follow`. When specified, only rows that satisfy the filter condition are included in the snapshot.
The condition must be a valid SQL Boolean expression that can be used in the `where` clause of the `select * from snapshot where ...` query.
Unlike the `filter` option, which applies to all records retrieved from the table, this filter only applies to rows in the initial snapshot of the table. For instance, it can be used to specify the range of event times to include in the snapshot, e.g.: `ts BETWEEN TIMESTAMP '2005-01-01 00:00:00' AND TIMESTAMP '2010-12-31 23:59:59'`. This option can be used together with the `filter` option. During the initial snapshot, only rows that satisfy both `filter` and `snapshot_filter` are retrieved from the Delta table. When subsequently following changes in the the transaction log (`mode = snapshot_and_follow`), all rows that meet the `filter` condition are ingested, regardless of `snapshot_filter`.
| | `version`, `start_version` | integer| |Optional table version. When this option is set, the connector finds and opens the specified version of the table. In `snapshot` and `snapshot_and_follow` modes, it retrieves the snapshot of this version of the table. In `follow`, `snapshot_and_follow`, and `cdc` modes, it follows transaction log records **after** this version.
Note: at most one of `version` and `datetime` options can be specified. When neither of the two options is specified, the latest committed version of the table is used.
| | `datetime` | string | |Optional timestamp for the snapshot in the ISO-8601/RFC-3339 format, e.g., "2024-12-09T16:09:53+00:00". When this option is set, the connector finds and opens the version of the table as of the specified point in time (based on the server time recorded in the transaction log, not the event time encoded in the data). In `snapshot` and `snapshot_and_follow` modes, it retrieves the snapshot of this version of the table. In `follow`, `snapshot_and_follow`, and `cdc` modes, it follows transaction log records **after** this version.
Note: at most one of `version` and `datetime` options can be specified. When neither of the two options is specified, the latest committed version of the table is used.
| @@ -359,6 +359,40 @@ it is marked **UNHEALTHY** while retrying failed operations. If the pipeline is stopped and restarted during a retry, the connector resumes from the last successfully ingested table version. This guarantees that no data loss occurs due to object store read errors. +## Optimizing multihost performance for unbalanced large data + +In a multihost pipeline, Feldera currently assigns each input +connector to one host. All of the work related to reading data from +that input connector then takes place on that host. When a pipeline +has several input connectors that read comparable amounts of data, +this spreads the network and CPU load related to them across the +hosts. + +On the other hand, if a pipeline that has one input connector that +reads most of the pipeline's data, only a single host does all the +work. In such a case, it makes sense to divide the connector into +multiple connectors, one per host. Feldera places input connectors on +hosts "round robin" in alphabetical order, which means that using the +default connector names or names with the same prefix and a sequential +suffix will ensure that they are spread as evenly as possible across +the hosts. + +When a Delta Lake source is divided into +[partitions](https://docs.delta.io/best-practices/), its input +connector can be divided into multiple connectors by specifying a +different [`filter`](#filter) for each one. For example, if the +source is partitioned by an integer column `partition` that takes one +of the six values 0 through 5, and there are 3 hosts, one might use 3 +copies of the input connector, one with `"filter": "partition = 0 OR +partition = 1"`, one with `"filter": "partition = 2 OR partition = +3"`, and one with `"filter": "partition = 4 OR partition = 5`. + +> ⚠️ [Tables with LATENESS] cannot correctly be divided into multiple +> connectors this way, because the different connectors can read data +> "out of sync" from one another. + +[Tables with LATENESS]: /sql/streaming/#lateness-expressions + ## Additional examples ### Example: Setting `timestamp_column` @@ -441,4 +475,4 @@ Read table snapshot via Unity catalog. "unity_host": "https://dbc-XXX-XXX.cloud.databricks.com", "aws_region": "us-west-1" } -``` \ No newline at end of file +``` diff --git a/docs.feldera.com/docs/connectors/sources/kafka.md b/docs.feldera.com/docs/connectors/sources/kafka.md index 2f2e8c6cff2..033992d5dda 100644 --- a/docs.feldera.com/docs/connectors/sources/kafka.md +++ b/docs.feldera.com/docs/connectors/sources/kafka.md @@ -13,7 +13,7 @@ tolerance](/pipelines/fault-tolerance). | `topic` (required) | string | | The Kafka topic to subscribe to. | | `bootstrap.servers` (required) | string | | A comma separated list of Kafka brokers to connect to. | `start_from` | variant | latest | The starting point for reading from the topic, one of `"earliest"`, `"latest"`, `{"offsets": [1, 2, 3, ...]}`, where `[1, 2, 3, ...]` are particular offsets within each partition in the topic, or `{"timestamp":The list of Kafka partitions to read from.
Only the specified partitions will be consumed. If this field is not set, the connector will consume from all available partitions.
If `start_from` is set to `offsets` and this field is provided, the number of partitions must exactly match the number of offsets, and the order of partitions must correspond to the order of offsets.
If offsets are provided for all partitions, this field can be omitted.
| +| `partitions` | integer list | |The list of Kafka partitions to read from.
Only the specified partitions will be consumed. If this field is not set, the connector will consume from all available partitions.
If `start_from` is set to `offsets` and this field is provided, the number of partitions must exactly match the number of offsets, and the order of partitions must correspond to the order of offsets.
If offsets are provided for all partitions, this field can be omitted.
| | `log_level` | string | | The log level for the Kafka client. | | `group_join_timeout_secs` | seconds | 10 | Maximum timeout (in seconds) for the endpoint to join the Kafka consumer group during initialization. | | `poller_threads` | positive integer | 3 | Number of threads used to poll Kafka messages. Setting it to multiple threads can improve performance with small messages. Default is 3. Each partition being read is assigned to one of the threads, so the connector automatically caps the thread count at the partition count. | @@ -23,7 +23,7 @@ tolerance](/pipelines/fault-tolerance). | `include_partition` | boolean | false | Whether to include Kafka partition in connector metadata (see [Accessing Kafka metadata](#metadata)). | | `include_offset` | boolean | false | Whether to include Kafka offset name in connector metadata (see [Accessing Kafka metadata](#metadata)). | | `include_timestamp` | boolean | false | Whether to include Kafka timestamp in connector metadata (see [Accessing Kafka metadata](#metadata)). | -| `synchronize_partitions` | boolean | false | Whether to read records in order of Kafka timestamp across partitions (see [Synchronizing partitions](#synchronizing-partitions)) | +| `synchronize_partitions` | boolean | false | Whether to read records in order of Kafka timestamp across partitions (see [Synchronizing partitions](#synchronizing-partitions)) | | `header_filter` | filter | | Drop messages whose Kafka headers do not match a boolean expression of regular expressions (see [Filtering messages by header](#header-filter)). | The connector passes additional options directly to [**librdkafka**](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md). Some of the relevant options: @@ -508,6 +508,40 @@ Pitfalls of this solution include: be added to the partition whose events have been completely processed. +## Optimizing multihost performance for unbalanced large data + +In a multihost pipeline, Feldera currently assigns each input +connector to one host. All of the work related to reading data from +that input connector then takes place on that host. When a pipeline +has several input connectors that read comparable amounts of data, +this spreads the network and CPU load related to them across the +hosts. + +On the other hand, if a pipeline that has one input connector that +reads most of the pipeline's data, only a single host does all the +work. In such a case, it makes sense to divide the connector into +multiple connectors, one per host. Feldera places input connectors on +hosts "round robin" in alphabetical order, which means that using the +default connector names or names with the same prefix and a sequential +suffix will ensure that they are spread as evenly as possible across +the hosts. + +A Kafka input connector can be divided into multiple connectors by +specifying different [`partitions`](#partitions) for each one. For +example, if the Kafka topic has 6 partitions, and there are 3 hosts, +one might use 3 copies of the input connector, one with `"partitions": +[0, 1]`, one with `"partitions": [2, 3]`, and one with `"partitions": +[4, 5]`. + +> ⚠️ The [synchronize_partitions] feature does not work across +> connectors, only within a single connector. Therefore, [tables with +> LATENESS] cannot correctly be divided into multiple connectors this +> way, because the different connectors can read data "out of sync" +> from one another. + +[tables with LATENESS]: /sql/streaming/#lateness-expressions +[synchronize_partitions]: #synchronize_partitions + ## Additional resources For more information, see: