> ## Documentation Index > Fetch the complete documentation index at: https://docs.risingwave.com/llms.txt > Use this file to discover all available pages before exploring further. # Ingest data from Pulsar > You can ingest data from Pulsar into RisingWave by using the Pulsar source connector in RisingWave. When creating a source, you can choose to persist the data from the source in RisingWave by using `CREATE TABLE` instead of `CREATE SOURCE` and specifying the connection settings and data format. ## Syntax ```sql theme={null} CREATE {TABLE | SOURCE} [ IF NOT EXISTS ] source_name [ schema_definition ] [INCLUDE { header | key | offset | partition | timestamp | payload } [AS ]] WITH ( connector='pulsar', connector_parameter='value', ... ) FORMAT data_format ENCODE data_encode ( message = 'message', schema.location = 'location' | schema.registry = 'schema_registry_url' ); ``` **schema\_definition**: ```sql theme={null} ( column_name data_type [ PRIMARY KEY ], ... [ PRIMARY KEY ( column_name, ... ) ] ) ``` For Avro and Protobuf data, do not specify `schema_definition` in the `CREATE SOURCE` or `CREATE TABLE` statement. The schema should be provided in a Web location in the option `schema.location` in `ENCODE properties` section. RisingWave performs primary key constraint checks on tables with connector settings but not on regular sources. If you need the checks to be performed, please create a table with connector settings. For a table with primary key constraints, if a new data record with an existing key comes in, the new record will overwrite the existing record. ### Connector parameters | Field | Notes | | :--------------------------------- | :------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | topic | **Required**. Address of the Pulsar topic. One source can only correspond to one topic. | | service.url | **Required**. Address of the Pulsar service. Typically in the format `pulsar://` or `pulsar+ssl://:` | | scan.startup.mode | **Optional**. The offset mode that RisingWave will use to consume data. The two supported modes are `earliest` (earliest offset) and `latest` (latest offset). If not specified, the default value `earliest` will be used. | | scan.startup.timestamp.millis. | **Optional**. RisingWave will start to consume data from the specified UNIX timestamp (milliseconds). | | auth.token | **Optional**. A token for auth. If both `auth.token` and `oauth` are set, only `oauth` authorization is effective. | | oauth.issuer.url | **Optional**. The issuer url for OAuth2. This field must be filled if other `oauth` fields are specified. | | oauth.credentials.url | **Optional**. The path for credential files, starts with `file://`. This field must be filled if other `oauth` fields are specified. | | oauth.audience | **Optional**. The audience for OAuth2. This field must be filled if other `oauth` fields are specified. | | oauth.scope | **Optional**. The scope for OAuth2. | | subscription.unacked.resend.delay | **Optional**. Specifies the delay duration after which the broker will resend unacknowledged messages. Accepts duration values such as `'1s'`, `'5m'`, or `'1h'`. Defaults to `None`, meaning the broker will **not** resend unacknowledged messages. | | pulsar.operation.retry.max.retries | **Optional**. Maximum number of retry attempts for retryable Pulsar client operations used by the source. If not specified, RisingWave preserves the Pulsar client's default retry behavior. | | pulsar.operation.retry.delay | **Optional**. Delay between retry attempts for retryable Pulsar client operations used by the source. Accepts duration values such as `'1s'`, `'5m'`, or `'1h'`. This is separate from `subscription.unacked.resend.delay`, which only controls broker redelivery of unacknowledged messages. | | pulsar.read\_compacted | **Optional**. Controls whether the consumer reads from the compacted topic or the full message backlog. When set to `true`, only the latest value for each key is read. When set to `false` or not specified, all messages are read. This option cannot be changed after the source or table is created. | ### Other parameters | Field | Notes | | :------------------------------------ | :--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | *data\_format* | Supported formats: DEBEZIUM, UPSERT, PLAIN. | | *data\_encode* | Supported encodes: JSON, AVRO, PROTOBUF, CSV, BYTES. | | *message* | Message name of the main Message in schema definition. Required when data\_encode is PROTOBUF. | | *location* | Web location of the schema file in `http://...`, `https://...`, or `S3://...` format. Required when `data_encode` is `AVRO` or `PROTOBUF`. Examples:`https://\/risingwave/proto-simple-schema.proto`,`s3://risingwave-demo/schema-location` | | *aws.credentials.access\_key\_id* | **Optional**. The AWS access key for loading from S3. This field does not need to be filled if `oauth.credentials.url` is specified to a local path. | | *aws.credentials.secret\_access\_key* | **Optional**. The AWS secret access key for loading from S3. This field does not need to be filled if `oauth.credentials.url` is specified to a local path. | | *region* | Required if loading descriptors from S3. The AWS service region. | | *aws.credentials.role.arn* | **Optional**. The Amazon Resource Name (ARN) of the role to assume. | | *aws.credentials.role.external\_id* | **Optional**. The [external](https://aws.amazon.com/blogs/security/how-to-use-external-id-when-granting-access-to-your-aws-resources/) id used to authorize access to third-party resources. | ## Message acknowledgment Added in v2.7.0. RisingWave automatically sends acknowledgments to the Pulsar broker for consumed messages. You can optionally configure the `subscription.unacked.resend.delay` parameter to control when the broker resends unacknowledged messages. If you need to bound retryable Pulsar client operations on the source side, configure `pulsar.operation.retry.max.retries` and `pulsar.operation.retry.delay`. These options control Pulsar client retries in RisingWave and do not change broker-side redelivery behavior. ```sql Example theme={null} CREATE TABLE my_pulsar_table ( f1 int, f2 varchar ) WITH ( connector = 'pulsar', service.url = 'pulsar://localhost:6650/', topic = 'my_topic', subscription.unacked.resend.delay = '10s' ) FORMAT PLAIN ENCODE JSON; ``` In this example, if the broker does not receive an acknowledgment within 10 seconds, it will resend the messages. ## Read from compacted topics Pulsar supports [topic compaction](https://pulsar.apache.org/docs/next/concepts-topic-compaction/), which retains only the latest message for each key in a topic. When you enable `pulsar.read_compacted`, RisingWave reads only the compacted view of the topic, providing the most recent value for each key. This is useful when you want to read the current state of entities without processing all historical updates. ```sql Example theme={null} CREATE TABLE user_status ( user_id int, status varchar ) WITH ( connector = 'pulsar', service.url = 'pulsar://localhost:6650/', topic = 'user_status_compacted', pulsar.read_compacted = 'true', scan.startup.mode = 'earliest' ) FORMAT PLAIN ENCODE JSON; ``` In this example, if multiple status updates exist for the same `user_id`, only the latest status for each user will be read from the compacted topic. The `pulsar.read_compacted` option cannot be changed after the source or table is created. ## Read schemas from locations RisingWave supports reading schemas from a Web location in `http://...`, `https://...`, or `S3://...` format for Pulsar data in Avro or Protobuf format. For Protobuf, if a schema location is specified, the schema file must be a `FileDescriptorSet`, which can be compiled from a `.proto` file with a command like this: ```bash theme={null} protoc -I=$include_path --include_imports --descriptor_set_out=schema.pb schema.proto ``` To specify a schema location, add this clause to a `CREATE SOURCE` statement. ```bash theme={null} ROW SCHEMA LOCATION 'location' ``` If a primary key also needs to be defined, use the table constraint syntax. ```sql theme={null} CREATE TABLE table1 (PRIMARY KEY(id)) ``` ## Example Here is an example of connecting RisingWave to a Pulsar broker to read data from individual topics. ```sql theme={null} CREATE {TABLE | SOURCE} IF NOT EXISTS source_abc WITH ( connector='pulsar', topic='demo_topic', service.url='pulsar://localhost:6650/', oauth.issuer.url='https://auth.streamnative.cloud/', oauth.credentials.url='s3://bucket_name/your_key_file.file', oauth.audience='urn:sn:pulsar:o-d6fgh:instance-0', aws.credentials.access_key_id='aws.credentials.access_key_id', aws.credentials.secret_access_key='aws.credentials.secret_access_key', scan.startup.mode='latest', scan.startup.timestamp.millis='140000000' ) FORMAT PLAIN ENCODE AVRO ( message = 'message', schema.location = 'https://demo_bucket_name.s3-us-west-2.amazonaws.com/demo.avsc' ); ``` ```sql theme={null} CREATE {TABLE | SOURCE} IF NOT EXISTS source_abc ( column1 varchar, column2 integer, ) WITH ( connector='pulsar', topic='demo_topic', service.url='pulsar://localhost:6650/', oauth.issuer.url='https://auth.streamnative.cloud/', oauth.credentials.url='s3://bucket_name/your_key_file.file', oauth.audience='urn:sn:pulsar:o-d6fgh:instance-0', aws.credentials.access_key_id='aws.credentials.access_key_id', aws.credentials.secret_access_key='aws.credentials.secret_access_key', scan.startup.mode='latest', scan.startup.timestamp.millis='140000000' ) FORMAT PLAIN ENCODE JSON; ``` Use the `payload` keyword to ingest JSON data when you are unsure of the exact schema beforehand. Instead of defining specific column names and types at the very beginning, you can load all JSON data first and then prune and filter the data during runtime. Check the example below: ```sql theme={null} CREATE TABLE table_include_payload (v1 int, v2 varchar) INCLUDE payload WITH ( connector = 'pulsar', topic = 'pulsar_1_partition_topic', properties.bootstrap.server = 'message_queue:29092', scan.startup.mode = 'earliest' ) FORMAT PLAIN ENCODE JSON; ``` ```sql theme={null} CREATE {TABLE | SOURCE} IF NOT EXISTS source_abc ( column1 varchar, column2 integer, ) WITH ( connector='pulsar', topic='demo_topic', service.url='pulsar://localhost:6650/', oauth.issuer.url='https://auth.streamnative.cloud/', oauth.credentials.url='s3://bucket_name/your_key_file.file', oauth.audience='urn:sn:pulsar:o-d6fgh:instance-0', aws.credentials.access_key_id='aws.credentials.access_key_id', aws.credentials.secret_access_key='aws.credentials.secret_access_key', scan.startup.mode='latest', scan.startup.timestamp.millis='140000000' ) FORMAT PLAIN ENCODE PROTOBUF ( message = 'package.message_name', schema.location = 'https://demo_bucket_name.s3-us-west-2.amazonaws.com/demo.proto' ); ```