@RunLLM postgres sink with JsonSchema

To use the Pulsar JDBC PostgreSQL sink with a JSON schema, you must define your schema as type JSON and upload it to the topic before creating the sink. The sink configuration should specify the JDBC connection details and table name. Only Avro and JSON schema types are supported for the sink, and the schema must match the table structure in PostgreSQL. For example, in the StreamNative Console, you can select "JSON" as the schema type and provide the schema definition accordingly. See the official docs for a full example and configuration steps: Pulsar IO Quickstart - Connect Pulsar to PostgreSQL. Would you like a step-by-step example or a sample configuration file? Sources: - Pulsar IO Quickstart - Connect Pulsar to PostgreSQL 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@RunLLM yes

Certainly! Here’s a step-by-step example of how to use the Pulsar JDBC PostgreSQL sink with a JSON schema. --- 1. Prepare PostgreSQL Start a PostgreSQL instance (for example, using Docker):

docker run -d --name pulsar-postgres -p 5432:5432 -e POSTGRES_PASSWORD=password -e POSTGRES_USER=postgres postgres:12

Create a table for your data:

-- Connect to PostgreSQL (e.g., psql -U postgres)
CREATE TABLE IF NOT EXISTS pulsar_postgres_jdbc_sink (
  id serial PRIMARY KEY,
  name VARCHAR(255) NOT NULL
);

--- 2. Define and Upload the JSON Schema Create a file named

json-schema.json

with your schema definition. For example:

{
  "type": "JSON",
  "schema": "{\"type\":\"record\",\"name\":\"Test\",\"fields\":[{\"name\":\"id\",\"type\":[\"null\",\"int\"]},{\"name\":\"name\",\"type\":[\"null\",\"string\"]}]}",
  "properties": {}
}

Upload the schema to your Pulsar topic:

bin/pulsar-admin schemas upload persistent://public/default/pulsar-postgres-jdbc-sink-topic -f ./json-schema.json

--- 3. Create the Sink Configuration Create a file named `pulsar-postgres-jdbc-sink.yaml`:

configs:
  userName: "postgres"
  password: "password"
  jdbcUrl: "jdbc:postgresql://localhost:5432/postgres"
  tableName: "pulsar_postgres_jdbc_sink"

--- 4. Create the JDBC Sink Run the following command to create the sink:

bin/pulsar-admin sinks create \
  --archive $PWD/connectors/pulsar-io-jdbc-postgres-<version>.nar \
  --inputs pulsar-postgres-jdbc-sink-topic \
  --name pulsar-postgres-jdbc-sink \
  --sink-config-file $PWD/pulsar-postgres-jdbc-sink.yaml \
  --parallelism 1

Replace

<version>

with the actual version of the connector you have. --- 5. Produce Messages with JSON Schema You can now produce messages to the topic using the JSON schema you defined. For example, using the Pulsar CLI:

bin/pulsar-client produce persistent://public/default/pulsar-postgres-jdbc-sink-topic \
  -m '{"id": 1, "name": "Alice"}'

--- 6. Verify Data in PostgreSQL Check your PostgreSQL table to see the ingested data:

SELECT * FROM pulsar_postgres_jdbc_sink;

--- References - Pulsar IO Quickstart - Connect Pulsar to PostgreSQL - JDBC Sink Connector Docs If you need a more advanced example (e.g., with nested fields or arrays), or if you want to see how to do this with StreamNative Cloud or the StreamNative Console, let me know! 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@RunLLM pulsar-client produce json

To produce JSON messages with the Pulsar CLI, use the

bin/pulsar-client produce

command and pass your JSON payload as a string with the

-m

or

--messages

flag. For example:

bin/pulsar-client produce my-topic --messages '{"id":1,"name":"Alice"}'

This sends a single JSON message to the topic. Make sure the topic has a compatible JSON schema if schema validation is enabled. For more, see Pulsar produce/consume tutorial. Would you like details on schema setup or producing multiple JSON messages? Sources: - Pulsar produce/consume tutorial 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@RunLLM Python example producer to send JSONschems

To send messages with a JSON schema using the Pulsar Python client, define a class inheriting from

pulsar.schema.Record

, then use

JsonSchema

when creating the producer. Example:

import pulsar
from pulsar.schema import *

class Example(Record):
    id = Integer()
    name = String()

client = pulsar.Client('pulsar://localhost:6650')
producer = client.create_producer(
    'my-topic',
    schema=JsonSchema(Example)
)

producer.send(Example(id=1, name='Alice'))
client.close()

See Python schema docs for more details. Would you like a consumer example or more on schema compatibility? Sources: - Python schema docs 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.