# CREATE STORE
## Syntax
```sql
CREATE STORE
store_name
WITH (store_parameter = value [, ...]);
```
## Description
DeltaStream processes streaming data that is stored in streaming stores such as Apache Kafka and Amazon Kinesis. The first step toward accessing this data is to connect to such data stores. You do this using the `CREATE STORE` statement.
{% hint style="info" %}
**Notes**
Currently DeltaStream supports Kafka (Confluent Cloud, Amazon MSK, RedPanda, and more) and Amazon Kinesis. Support for additional streaming stores such as Google Pub/Sub and Apache Pulsar are coming soon.\
\
DeltaStream also provides non-streaming store support for ClickHouse, DataBricks, Iceberg AWS Glue Catalog, Iceberg REST Catalog, S3, and PostgreSQL. \
\
Each non-streaming store has a different set of limitations, listed below.
{% endhint %}
Only a [role](https://docs.deltastream.io/overview/core-concepts/access-control#_role) with [`CREATE_STORE`](https://docs.deltastream.io/overview/core-concepts/access-control#privilege) privilege can create a store.
### Arguments
#### store\_name
Name of the store to define. If the name is case sensitive you must wrap it in double quotes; otherwise the system uses the lower case name.
#### WITH (store\_parameter = value \[, … ])
This clause specifies [#store\_parameters](#store_parameters "mention").
### Store Parameters
| Parameter Name | Description |
|---|
type | Specifies the store type.
Required: Yes
Type: STORE_TYPE Valid values: KAFKA, KINESIS, SNOWFLAKE, DATABRICKS, POSTGRESQL |
uris | List of comma-separated host:port URIs to connect to the store.
Required: Yes, unless specified in properties.file.
Type: String |
tls.disabled | Optional. Specifies if the store should be accessed over TLS.
Required: No
Default value: FALSE Type: Boolean Valid values: TRUE or FALSE |
tls.verify_server_hostname | Specifies if the server CNAME should be validated against the certificate.
Required: No
Default value: TRUE
Type: Boolean
Valid values: TRUE or FALSE |
tls.ca_cert_file | Path to a CA certificate file in PEM format. Preface the path with @ for the file to be uploaded to the server.
Required: No
Default value: Public CA chains.
Type: String
Valid values: Path to a SSL certificate in PEM format |
tls.cipher_suites | Comma-separated list of cipher suites to use when establishing a TLS connection.
Required: No
Default value: [] (all supported cipher suites are enabled)
Type: List
Valid values: Full cipher suite names describing algorithm content |
tls.protocols | Comma-separated list TLS protocol versions to use while establishing a TLS connection.
Required: No
Default value: TLSv1.2,TLSv1.1,TLSv1
Type: List
Valid values: TLS protocols with version |
schema_registry.name | Name of a schema registry to associate with the store. You must first create a schema registry using the create-schema_registry DDL statement. Only one schema registry can be associated with a store.
Required: No
Default value: None
Type: String
Valid values: See list-schema_registries |
properties.file | The file path to a .yaml file containing any store parameter. Preface the path with @ for the file to be uploaded to the server.
Required: No
Default value: None
Type: String
Valid values: File path in current user's filesystem |
### **ClickHouse-Specific Parameters**
Parameters to be used if `type` is `CLICKHOUSE`:
| Parameter Name | Description |
| --------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `clickhouse.username` | Username to connect to the database instance specified with the store's uris parameter.
Required: Yes
Default value: None
Type: String
|
| `clickhouse.password` | Password to connect to the database instance using the store's username parameter.
Required: Yes
Default value: None
Type: String
|
### **Databricks-Specific Parameters**
Parameters to be used if `type` is `DATABRICKS`:
| Parameter Name | Description |
|---|
databricks.app_token | Databricks personal access token used when authenticating with a Databricks workspace.
Required: Yes
Default value: None
Type: String |
databricks.warehouse_id | The identifier for a Databricks SQL warehouse belonging to a Databricks workspace. This warehouse is used to create and query tables in Databricks.
Required: Yes Default value: None
Type: String |
databricks.warehouse_port | The port for a Databricks SQL warehouse belonging to a Databricks workspace.
Required: No Default value: 443
Type: Integer |
aws.access_key_id | AWS access key ID used for writing data to S3.
Required: Yes Default value: None
Type: String |
aws.secret_access_key | AWS secret access key used for writing data to S3.
Required: Yes
Default value: None
Type: String
|
databricks.cloud.s3.bucket | The AWS S3 bucket that create-table-as queries will write data to.
Required: Yes
Default value: None
Type: String |
databricks.cloud.region | The cloud region that the databricks.cloud.s3.bucket belongs to.
Required: Yes
Default value: None
Type: String Valid values: AWS us-east-1AWS us-east-2AWS us-west-1AWS us-west-2
|
### **Iceberg REST Catalog-Specific Parameters**
Parameters to be used if `type` is `ICEBERG REST`:
| Parameter Name | Description |
| ---------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `iceberg.rest.client_id` | Iceberg client\_id
Required: Yes
Default value: NONE
Type: STRING
|
| `iceberg.rest.client_secret` | Iceberg client\_secret
Required: Yes
Default value: NONE
Type: STRING
|
| `iceberg.rest.client_scope` | Iceberg client\_scope
Required: Yes
Default value: NONE
Type: STRING
|
| `iceberg.catalog.id` | Iceberg catalog name
Required: Yes
Default value: NONE
Type: STRING
|
### **Kafka-Specific Parameters**
Parameters to be used if `type` is `KAFKA`:
| Parameter Name | Description |
|---|
kafka.sasl.hash_function | SASL hash function to use when authenticating with Apache Kafka brokers.
Required: No
Default value: NONE.
Type: HASH_FUNCTION
Valid values: NONE, PLAIN, SHA256, SHA512, and AWS_MSK_IAM |
kafka.sasl.username | Username to use when authenticating with Apache Kafka brokers.
Required: Yes, if kafka.sasl.hash_function is not NONE or AWS_MSK_IAM Default value: None
Type: String |
kafka.sasl.password | Password to use when authenticating with Apache Kafka brokers.
Required: Yes, if kafka.sasl.hash_function is not NONE or AWS_MSK_IAM Default value: None
Type: String |
kafka.msk.aws_region | AWS region to use when authenticating with MSK. Required: Yes, if kafka.sasl.hash_function is AWS_MSK_IAM
Default value: None
Type: String
Example: us-east-1 |
kafka.msk.iam_role_arn | AWS IAM role ARN to use when authenticating with MSK. Required: Yes, if kafka.sasl.hash_function is AWS_MSK_IAM
Default value: None
Type: String
Example: arn:aws:iam::123456789012:role/example-IAM-role |
tls.client.cert_file | Path to a client certificate file in PEM format. Preface the path with @ for the file to be uploaded to the server.
Required: Yes, if kafka.sasl.hash_function is SHA256 or SHA512
Default value: None
Type: String
Valid value: Path to a SSL certificate in PEM format |
tls.client.key_file | Path to the client key file in PEM format. Prefix the path with @ for the file to be uploaded to the server.
Required: Yes, if kafka.sasl.hash_function is SHA256 or SHA512
Default value: None
Type: String
Valid value: Path to a SSL certificate in PEM format |
### **Kinesis-Specific Parameters**
Parameters to be used if `type` is `KINESIS`:
| Parameter Name | Description |
|---|
kinesis.iam_role_arn | AWS IAM role ARN to use when authenticating with an Amazon Kinesis service. Required: Yes, unless authenticating with the Amazon Kinesis Service using static AWS credentials
Default value: None
Type: String
Example: arn:aws:iam::123456789012:role/example-IAM-role |
kinesis.access_key_id | AWS IAM access key to use when authenticating with an Amazon Kinesis service.
Required: Yes, if authenticating with the Amazon Kinesis Service using static AWS credentials
Default value: None
Type: String |
kinesis.secret_access_key | AWS IAM secret access key to use when authenticating with an Amazon Kinesis service.
Required: Yes, if authenticating with the Amazon Kinesis Service using static AWS credentials
Default value: None
Type: String |
### PostgreSQL-Specific Parameters
Parameters to be used if `type` is `PostgreSQL`:
| Parameter Name | Description |
| ---------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `postgres.username` | Username to connect to the database instance specified with the store's uris parameter.
Required: Yes, if using user/pass authentication
Default value: None
Type: String
|
| `postgres.password` | Password to connect to the database instance using the store's username parameter.
Required: Yes, if using user/pass authentication
Default value: None
Type: String
|
| `tls.client.cert_file` | Path to a client certificate file in PEM format. Preface the path with @ for the file to be uploaded to the server.
Required: Yes, if using TLS client certificate- based authentication
Default value: None
Type: String
Valid value: Path to a SSL certificate in PEM format
|
| `tls.client.key_file` | Path to the client key file in PEM format. Prefix the path with @ for the file to be uploaded to the server.
Required: Yes, if using TLS client certificate based authentication
Default value: None
Type: String
Valid value: Path to a SSL certificate in PEM format
|
{% hint style="warning" %}
To secure the Postgres connection, DeltaStream requires you to set `tls.verify_server_hostname`to `TRUE` and to set `tls.disabled` to `FALSE` .
{% endhint %}
#### Using Neon-Hosted PostgreSQL
If your PostgreSQL instance is hosted on [Neon](https://neon.com/), you must make the following adjustments to the Store URI at creation time for it to work properly:
1. **Explicitly include the port number in the URI.** Neon uses the default PostgreSQL port 5432, but it must be specified explicitly. Check [here](https://neon.com/docs/connect/query-with-psql-editor#what-port-does-neon-use) for details.
2. **Add the `endpoint` as a connection option** **in the Store URI.**
**Example**
```
// Neon connection URI (as provided by Neon):
postgresql://.c-3.us-east-1.aws.neon.tech/
// Store URI used when creating the Store
postgresql://.c-3.us-east-1.aws.neon.tech:5432/?options=endpoint%3D
```
### S3-Specific Parameters
For an S3 store, the `uris` parameter must specify the S3 URI of a bucket or folder whose contents form the data source for the Store. Other parameters to be used if `type` is `S3` are:
| Parameter Name | Description |
| ----------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| `aws.iam_role_arn` | AWS IAM role ARN to use when authenticating with S3.
Required: No
Default value: None
Type: String
|
| `aws.iam_external_id` | IAM External ID.
Required: Yes, if aws.iam\_role\_arn is specified.
Default value: None
Type: String
|
| `aws.access_key_id` | AWS IAM role ARN to use when authenticating with S3.
Required: Yes, if authenticating using static AWS credentials.
Default value: None
Type: String
|
| `aws.secret_access_key` | IAM External ID.
Required: Yes, if authenticating using static AWS credentials.
Default value: None
Type: String
|
### Setting a Trust Relationship when Using Your IAM Role ARN and External ID
Before you can access a data store using `aws.iam_role_arn` and `aws.iam_external_id`, you must update the role provided in `aws.iam_role_arn` for a trust relationship to include a policy for a DeltaStream Platform Role. The DeltaStream Platform Role attempts to assume the provided role.\
\
The trust relationship JSON resembles the following:
```
{
"Sid": "DeltaStreamTrustPolicy",
"Effect": "Allow",
"Principal": {
"AWS": [
"arn:aws:iam::DELTASTREAM_DATAPLANE_AWS_ACCOUNT:role/DELTASTREAM_CROSS_ACCOUNT_IAM_ROLE_ARN",
]
},
"Action": "sts:AssumeRole",
"Condition": {
"StringEquals": {
"sts:ExternalId": "EXTERNAL_ID"
}
}
}
```
In the trust policy you must substitute following parameters:
* DELTASTREAM_DATAPLANE_AWS_ACCOUNT - Reach out to DeltaStream for the dataplane AWS Account
* DELTASTREAM_CROSS_ACCOUNT_IAM_ROLE_ARN - Reach out to DeltaStream for dataplane-specific IAM Role ARN
* ```
EXTERNAL_ID - This should match the provided aws.iam_external_id
```
For further details on how to configure a trust relationship, please refer to the AWS guide:
### Snowflake-Specific Parameters
Parameters to be used if `type` is `SNOWFLAKE`:
| Parameter Name | Description |
|---|
snowflake.account_id | Snowflake account identifier assigned to the Snowflake account.
Required: Yes
Default value: None
Type: String |
snowflake.cloud.region | Name of the Snowflake cloud region in which the account resources operate.
Required: Yes Default value: None
Type: String
Valid values: AWS us-east-1AWS us-east-2AWS us-west-1AWS us-west-2
|
snowflake.role_name | Access control role to use for the store operations after connecting to Snowflake.
Required: Yes Default value: None
Type: String |
snowflake.username | User login name for the Snowflake account.
Required: Yes
Default value: None
Type: String |
snowflake.warehouse_name | Warehouse name to use for queries and other store operations that require compute resources.
Required: Yes
Default value: None
Type: String |
snowflake.client.key_file | Path to the Snowflake account's private key in PEM format. Prefix the path with @ for the file to be uploaded to the server.
Required: Yes
Default value: None
Type: String |
snowflake.client.key_passphrase | Passphrase for decrypting the Snowflake account's private key.
Required: No
Default value: None
Type: String |
## Examples
#### Create a Kafka store with credentials
The following creates a new Kafka store with name `my_kafka_store`:
```sql
CREATE STORE
my_kafka_store
WITH (
'type' = KAFKA,
'uris' = 'kafka.broker1.url:9092,kafka.broker2.url:9092',
'tls.ca_cert_file' = '@/certs/us-east-1/self-signed-kafka-ca.crt'
);
```
#### Create an MSK store with IAM credentials
The following creates a new Kafka store with name `my_kafka_store`:
```sql
CREATE STORE
msk_iam_store
WITH (
'type' = KAFKA,
'uris' = 'b-1.abc.abc.c10.kafka.us-east-1.amazonaws.com:9098,b-2.abc.abc.c10.kafka.us-east-1.amazonaws.com:9098,b-3.abc.abc.c10.kafka.us-east-1.amazonaws.com:9098',
'kafka.sasl.hash_function' = AWS_MSK_IAM,
'kafka.msk.aws_region'='us-east-1',
'kafka.msk.iam_role_arn'='arn:aws:iam::123456789012:role/example-IAM-role'
);
```
#### Create a Kafka store with credentials from a file
The following creates a new Kafka store with name `MyKafkaStore++`:
```sql
CREATE STORE
"MyKafkaStore++"
WITH (
'type' = KAFKA,
'properties.file' = '@/User/user1/deltastream/kafka_store/properties.yaml'
);
```
```sh
$ cat /User/user1/deltastream/kafka_store/properties.yaml
uris: "http://uri1,http://uri2"
kafka.sasl.hash_function: PLAIN
kafka.sasl.username: "ABCDEFGH12345678"
kafka.sasl.password: "kafkasaslpassword"
```
#### Create a Kinesis store with IAM credentials
The following statement creates a new Kinesis store with name `my_kinesis_store`:
```sql
CREATE STORE
my_kinesis_store
WITH (
'type' = KINESIS,
'uris' = 'https://url.to.kinesis.aws:4566',
'kinesis.iam_role_arn' = 'arn:aws:iam::123456789012:role/example-IAM-role'
);
```
#### Create a Kinesis store with static credentials
The following statement creates a new Kinesis store with name `my_kinesis_store`:
```sql
CREATE STORE
my_kinesis_store
WITH (
'type' = KINESIS,
'uris' = 'https://url.to.kinesis.aws:4566',
'kinesis.access_key_id' = 'testkey',
'kinesis.secret_access_key' = 'testsecret'
);
```
#### Create a Kafka store with a schema registry
The following statement creates a new Kafka store with a schema registry named `sr`. Note that the store name is case-sensitive and thus has quotes around it:
```sql
CREATE STORE
"kafkaStoreWithSR"
WITH (
'type' = KAFKA,
'uris' = 'kafka.broker1.url:9092,kafka.broker2.url:9092',
'schema_registry.name' = sr
);
```
#### Create a Confluent Kafka store with credentials
The following creates a new Confluent Cloud Kafka store with the case-sensitive name `ConfluentCloudKafkaStore`:
```sql
CREATE STORE "ConfluentCloudKafkaStore"
WITH (
'type' = KAFKA,
'uris' = 'abc-12345.us-east-1.aws.confluent.cloud:9092',
'kafka.sasl.hash_function' = PLAIN,
'kafka.sasl.username' = 'credentials_username',
'kafka.sasl.password' = 'credentials_password'
);
```
#### Create a Snowflake store
```sql
CREATE STORE sf
WITH (
'type' = SNOWFLAKE,
'uris' = 'https://my-account.snowflakecomputing.com',
'snowflake.account_id' = 'my-account',
'snowflake.role_name' = 'ACCOUNTADMIN',
'snowflake.username' = 'STREAMING_USER',
'snowflake.warehouse_name' = 'COMPUTE_WH',
'snowflake.client.key_file' = '@/path/to/pk/my_account_rsa.p8'
);
```
#### Create a Snowflake store with client key passphrase
```sql
CREATE STORE sf
WITH (
'type' = SNOWFLAKE,
'uris' = 'https://my-account.snowflakecomputing.com',
'snowflake.account_id' = 'my-account',
'snowflake.role_name' = 'ACCOUNTADMIN',
'snowflake.username' = 'STREAMING_USER',
'snowflake.warehouse_name' = 'COMPUTE_WH',
'snowflake.client.key_file' = '@/path/to/pk/my_account_rsa.p8',
'properties.file' = '@/path/to/deltastream/snowflake_store/properties.yaml'
;
```
```sh
$ cat /path/to/deltastream/snowflake_store/properties.yaml
snowflake.client.key_passphrase: "my$account$$key$$$phrase"
```
#### Create a Databricks store
```sql
CREATE STORE databricks_store WITH (
'type' = DATABRICKS,
'uris' = 'https://dbc-12345678-1234.cloud.databricks.com',
'databricks.app_token' = 'dapiabcdefghijklmnopqrstuvw123456789',
'databricks.warehouse_id' = 'abcdefgh1234567',
'aws.access_key_id' = 'AWS_ACCESS_KEY',
'aws.secret_access_key' = 'AWS_SECRET_ACCESS_KEY',
'databricks.cloud.s3.bucket' = 'mybucket',
'databricks.cloud.region' = 'AWS us-west-2'
);
```
#### Create a PostgreSQL store
```sql
CREATE STORE ps_store WITH (
'type' = POSTGRESQL,
'uris' = 'postgresql://mystore.com:5432/demo',
'postgres.username' = 'user',
'postgres.password' = 'password'
);
```
#### Create a ClickHouse store
```sql
CREATE STORE ch WITH (
'type' = CLICKHOUSE,
'uris' = 'jdbc:clickhouse://uri:8443',
'clickhouse.username' = 'my_user',
'clickhouse.password' = 'my_pass'
);
```
#### Create an Iceberg AWS Glue Catalog Store with IAM credentials
```sql
CREATE STORE iceberg_glue_store WITH (
'type' = ICEBERG_GLUE,
'uris' = 'https://mybucket.s3.amazonaws.com/',
'aws.iam_role_arn'='arn:aws:iam::123456789012:role/example-IAM-role',
'aws.iam_external_id'='EXTERNAL ID'
'aws.region'='us-east-1',
'iceberg.warehouse.default_path'='s3://bucket/path',
'iceberg.catalog.id'='mycatalog'
);
```
#### Create an Iceberg REST Catalog Store using Snowflake OpenCatalog (Apache Polaris)
```sql
CREATE STORE iceberg_rest_store WITH (
'type' = ICEBERG_REST,
'uris' = 'https://abcd-xyz.snowflakecomputing.com/polaris/api/catalog',
'iceberg.catalog.id' = 'my-catalog',
'iceberg.rest.client_id' = 'CLIENT_ID',
'iceberg.rest.client_secret' = 'CLIENT_SECRET',
'iceberg.rest.scope' = 'SCOPE'
);
```
#### Create an S3 store
```sql
CREATE STORE s3_store WITH (
'type' = S3,
'uris' = 'https://mybucket.s3.amazonaws.com/',
'aws.access_key_id'='AWS_ACCESS_KEY',
'aws.secret_access_key'='AWS_SECRET_ACCESS_KEY'
);
```
#### Create an S3 Store with IAM credentials
```sql
CREATE STORE s3_store WITH (
'type' = S3,
'uris' = 'https://mybucket.s3.amazonaws.com/',
'aws.iam_role_arn'='arn:aws:iam::123456789012:role/example-IAM-role',
'aws.iam_external_id'='EXTERNAL ID'
);
```
**Notes**
All parameters above are required