CosmosDB
ProphecyWarehousePython0.0.1+Databricks UC Single ClusterNot SupportedDatabricks UC SharedNot SupportedLivyNot Supported
The azure-cosmos-spark connector is an integration between Azure CosmosDB and Apache Spark, enabling you to read and write data from and to CosmosDB using Spark. This document will cover the key properties available for configuration when using this connector.
Installation
On your Execution Cluster (Databricks or OnPrem), connect a dependency that has the following Maven coordinates (for Spark 3.3; For others please check this link):
com.azure.cosmos.spark:azure-cosmos-spark_3-3_2-12:4.18.1
Usage
Common Auth Properties
These properties are shared between the read and write operations.
| Property | Required | Default Value | Description |
|---|---|---|---|
| Authentication Type | True | MasterKey | There are two auth types are supported currently: MasterKey(PrimaryReadWriteKeys, SecondReadWriteKeys, PrimaryReadOnlyKeys, SecondReadWriteKeys), ServicePrinciple. |
| Azure Environment | True | Azure | The azure environment of the CosmosDB account: Azure, AzureChina, AzureUsGovernment, AzureGermany. |
| Account Endpoint | True | None | The CosmosDB account URI. |
| Account Key | True | None | The CosmosDB account key. |
| Subscription ID | False | None | The subscriptionId of the CosmosDB account. Required for ServicePrinciple authentication. |
| Tenant | False | None | The tenantId of the CosmosDB account. Required for ServicePrinciple authentication. |
| Resource Group Name | False | None | The resource group of the CosmosDB account. Required for ServicePrinciple authentication. |
| Client Id | False | None | The clientId/ApplicationId of the service principle. Required for ServicePrinciple authentication. |
| Client Secret | False | None | The client secret/password of the service principle. Required for ServicePrinciple authentication. |
| Database | True | None | The CosmosDB database name. |
Source Properties
Important Source Properties
These properties are specific to read operations.
| Property | Required | Default Value | Description |
|---|---|---|---|
| Data Source | True | DB Table | Whether all data in container should be loaded or should data be returned using a custom query |
| Container | False | None | The CosmosDB container name. |
| Custom Query | False | SELECT 1 | When provided the custom query will be processed against the CosmosDB endpoint instead of dynamically generating the query via predicate push down. Usually it is recommended to rely on Spark's predicate push down because that will allow to generate the most efficient set of filters based on the query plan. But there are a couple of predicates like aggregates (count, group by, avg, sum, etc.) that cannot be pushed down yet (at least in Spark 3.1) - so the custom query is a fallback to allow them to be pushed into the query sent to Cosmos. If specified, with schema inference enabled, the custom query will also be used to infer the schema. |
Other Source Properties
| Property | Required | Default Value | Description |
|---|---|---|---|
| Use Gateway Mode | False | false | Use gateway mode for the client operations. |
| Force Eventual Consistency | False | true | Makes the client use Eventual consistency for read operations instead of using the default account level consistency. |
| Application name | False | None | Application name |
| Preferred Regions List | False | None | Preferred regions list to be used for a multi region CosmosDB account. This is a comma separated value (e.g., [East US, West US] or East US, West US) provided preferred regions will be used as hint. You should use a collocated Spark cluster with your CosmosDB account and pass the Spark cluster region as preferred region. See list of azure regions here. Note that you can also use spark.cosmos.preferredRegions as alias. |
| Disable Tcp Connection Endpoint Rediscovery | False | false | Can be used to disable TCP connection endpoint rediscovery. TCP connection endpoint rediscovery should only be disabled when using custom domain names with private endpoints when using a custom Spark environment. When using Azure Databricks or Azure Synapse as Spark runtime it should never be required to disable endpoint rediscovery. |
| Allow Invalid JSON With Duplicate JSON Properties | False | false | By default (when set to false) the CosmosDB Java SDK and Spark connector will raise a hard failure when JSON documents are read that contain JSON object with multiple properties of the same name. This config option can be used to override the behavior and silently ignore the invalid JSON and instead use the last occurrence of the property when parsing the JSON. NOTE: This is only meant to be used as a temporary workaround. We strongly recommend fixing the invalid JSON from even being ingested into the data and only use this workaround while cleaning up the documents with invalid JSON. |
| Max Item Count | False | 1000 | Overrides the maximum number of documents that can be returned for a single query or change feed request. The default value is 1000. Consider increasing this only for average document sizes significantly smaller than 1KB or when projection reduces the number of properties selected in queries significantly (like when only selecting "id" of documents, etc.). |
| Max Integrated Staleness in milliseconds | False | None | Sets the max staleness window in milliseconds for the point read or query request results in the integrated cache when using the dedicated gateway. Learn more about max integrated cache staleness here. |
| Inclusion Mode | False | Always | Determines whether null/default values will be serialized to JSON or whether properties with null/default value will be skipped. The behavior follows the same ideas as Jackson's JsonInclude.Include. Always means JSON properties are created even for null and default values. NonNull means no JSON properties will be created for explicit null values. NonEmpty means JSON properties will not be created for empty string values or empty arrays/mpas. NonDefault means JSON properties will be skipped not just for null/empty but also when the value is identical to the default value 0 for numeric properties for example. |
| DateTime Conversion Mode | False | Default | The date/time conversion mode options are Default, AlwaysEpochMilliseconds, AlwaysEpochMillisecondsWithSystemDefaultTimezone. With Default, the standard Spark 3.* behavior is used: java.sql.Date/java.time.LocalDate are converted to EpochDay, java.sql.Timestamp/java.time.Instant are converted to MicrosecondsFromEpoch. With AlwaysEpochMilliseconds, the same behavior as the CosmosDB connector for Spark 2.4 is used: java.sql.Date, java.time.LocalDate, java.sql.Timestamp and java.time.Instant are converted to MillisecondsFromEpoch. The behavior for AlwaysEpochMillisecondsWithSystemDefaultTimezone is identical with AlwaysEpochMilliseconds except that the connector will assume System default time zone / Spark session time zone (specified via spark.sql.session.timezone) instead of UTC when the date/time to be parsed has no explicit time zone. |
| Schema Conversion Mode | False | Relaxed | The schema conversion behavior (Relaxed, Strict). When reading JSON documents, if a document contains an attribute that does not map to the schema type, the user can decide whether to use a null value (Relaxed) or an exception (Strict). |
| Partitioning Strategy | False | Default | The partitioning strategy used (Default, Custom, Restrictive or Aggressive) |
| Partitioning Targeted Count | False | None | An Integer value representing the targeted Partition Count. This parameter is optional and ignored unless strategy is Custom. In this case the Spark Connector won't dynamically calculate the number of partitions but stick with this value. |
| Partitioning Feed Range Filter | False | None | Can be used to scope the query to a single logical CosmosDB partition (or a subset of logical partitions). If this parameter is optionally provided, the partitioning strategy will be modified - only partitions for the scoped logical partitions will be created. So, the main benefit of this config option is to reduce the necessary SparkTasks/Partitions. |
Schema Inference Related Configs
| Property | Required | Default Value | Description |
|---|---|---|---|
| Enable Infer Schema | True | true | Whether all data in the container should be loaded or should data be returned using a custom query |
| Enable Custom Query for Inferring Schema | False | None | Checkbox, which if marked, will open up an editor for writing Custom SQL Query |
| Make all Columns Nullable | False | true | When schema inference is enabled, whether the resulting schema will make all columns nullable. By default, all columns (except cosmos system properties) will be treated as nullable even if all rows within the sample set have non-null values. When disabled, the inferred columns are treated as nullable or not depending on whether any record in the sample set has null-values within a column. |
| Record Sample Size for Schema Inference | False | 1000 | Sampling size to use when inferring schema and not using a query. |
| Include all System Properties | False | false | When schema inference is enabled, whether the resulting schema will include all CosmosDB system properties. |
| Include Document Timestamp field | False | false | When schema inference is enabled, whether the resulting schema will include the document Timestamp (_ts). Not required if spark.cosmos.read.inferSchema.includeSystemProperties is enabled, as it will already include all system properties. |
| Infer Schema Query | False | SELECT * FROM r | When schema inference is enabled, used as custom query to infer it. For example, if you store multiple entities with different schemas within a container and you want to ensure inference only looks at certain document types or you want to project only particular columns. |
Throughput Control Configs
Having throughput control helps to isolate the performance needs of applications running against a container, by limiting the amount of request units that can be consumed by a given Spark client. There are several advanced scenarios that benefit from client-side throughput control:
- Different operations and tasks have different priorities - there can be a need to prevent normal transactions from being throttled due to data ingestion or copy activities. Some operations and/or tasks aren't sensitive to latency, and are more tolerant to being throttled than others.
- Provide fairness/isolation to different end users/tenants - An application will usually have many end users. Some users may send too many requests, which consume all available throughput, causing others to get throttled.
- Load balancing of throughput between different Azure Cosmos DB clients - in some use cases, it's important to make sure all the clients get a fair (equal) share of the throughput.
| Property | Required | Default Value | Description |
|---|---|---|---|
| Throughput Control Enabled | False | false | Whether throughput control is enabled |
| Throughput Control: Account Endpoint | False | None | CosmosDB Account Endpoint Uri for throughput control. If not defined, then spark.cosmos.accountEndpoint will be used. |
| Throughput Control Account Key | False | None | CosmosDB Account Key for throughput control. |
| Throughput Control Preferred Regions List | False | None | Preferred regions list to be used for a multi region CosmosDB account. This is a comma separated value (e.g., [East US, West US] or East US, West US) provided preferred regions will be used as hint. You should use a collocated Spark cluster with your CosmosDB account and pass the Spark cluster region as preferred region. See list of azure regions here. |
| Disable TCP connection endpoint Rediscovery | False | false | Can be used to disable TCP connection endpoint rediscovery. TCP connection endpoint rediscovery should only be disabled when using custom domain names with private endpoints when using a custom Spark environment. When using Azure Databricks or Azure Synapse as Spark runtime it should never be required to disable endpoint rediscovery. |
| Use Gateway Mode | False | false | Use gateway mode for the client operations |
| Use Dedicated Container | False | true | Flag to indicate when configured with throughput control, whether dedicated throughput control container will be provided. |
| Throughput control group name | False | None | Throughput control group name |
| Throughput control group target throughput | False | None | Throughput control group target throughput in number of request units |
| Throughput control group target throughput threshold | False | None | Throughput control group target throughput threshold in request units |
| Database which will be used for throughput global control | False | None | Database which will be used for throughput global control |
| Container which will be used for throughput global control | False | None | Container which will be used for throughput global control |
| Renew Interval in milliseconds | False | 5s | How often the client is going to update the throughput usage of itself |
| Expire Interval in milliseconds | False | 11s | How quickly an offline client will be detected |
More information about Throughput control is available here.
Target Properties
These properties are specific to write operations.
Write Configurations
| Property | Required | Default Value | Description |
|---|---|---|---|
| Write Mode | True | append | Spark Write Mode. The possible values are error, append, overwrite and ignore . More information on Write Modes is available here. |
| Date Format | False | yyyy-MM-dd | Format for Date columns. More information on the possible formats is given here. |
| Timestamp Format | False | yyyy-MM-dd'T'HH:mm:ss.SSSXXX | Format for Timestamp columns. More information on the possible formats is given here. |
| Write Strategy | True | ItemOverwrite | CosmosDB Item Write Strategy: ItemOverwrite (using upsert); ItemOverwriteIfNotModified (if etag property of the row is empty/null it will just do an insert and ignore if the document already exists - same as ItemAppend, if an etag value exists the connector will attempt to replace the document with etag pre-condition. If the document changed - identified by precondition failure - the update is skipped and the document is not updated with the content of the data frame row), ItemAppend (using create, ignore pre-existing items i.e., Conflicts), ItemDelete (delete all documents), ItemDeleteIfNotModified (delete all documents for which the etag hasn't changed),ItemPatch (Partial update all documents based on the patch config). |
| Max Retry Attempts | False | 10 | CosmosDB Write Max Retry Attempts on retriable failures (e.g., connection error, write error, etc.) |
| Max Concurrency | False | None | CosmosDB Item Write Max concurrency. If not specified it will be determined based on the Spark executor VM Size. |
| Max No. of Pending Bulk Operations | False | None | CosmosDB Item Write bulk mode maximum pending operations. Defines a limit of bulk operations being processed concurrently. If not specified it will be determined based on the Spark executor VM Size. If the volume of data is large for the provisioned throughput on the destination container, this setting can be adjusted by following the estimation of 1000 x Cores. |
| Enable Write Bulk | False | true | CosmosDB Item Write bulk enabled |
Patch Specific Configurations
These Configurations are valid only when Write Strategy is given as ItemPatch
| Property | Required | Default Value | Description |
|---|---|---|---|
| Default Patch Operation Type | True | Replace | Default CosmosDB patch operation type. Supported types include none, add, set, replace, remove, increment. Choose none for no-op, for others please reference here for full context. |
| Patch Column Configs | False | None | CosmosDB patch column configs. It can contain multiple definitions that match the following patterns (separated by comma) => col(column).op(operationType) or col(column).path(patchInCosmosdb).op(operationType) - The difference of the second pattern is that it also allows you to define a different CosmosDB path. |
| Patch Filter | False | None | Used for Conditional patch |