versioned_docs/version-4.2.x/io-quickstart.md

---
id: io-quickstart
title: How to connect Pulsar to database
sidebar_label: "Get started"
description: Get started to connect Pulsar to database.
---

This tutorial provides a hands-on look at how you can move data out of Pulsar without writing a single line of code.

It is helpful to review the [concepts](io-overview.md) for Pulsar I/O by running the steps in this guide to gain a deeper understanding.

At the end of this tutorial, you can:

- [Connect Pulsar to Cassandra and ScyllaDB](#connect-pulsar-to-cassandra-and-scylladb)
- [Connect Pulsar to PostgreSQL](#connect-pulsar-to-postgresql)

:::tip

* These instructions assume you are running Pulsar in standalone mode. However, all the commands used in this tutorial can be used in a multi-node Pulsar cluster without any changes.
* All the instructions are assumed to run at the root directory of a Pulsar binary distribution.

:::

## Install Pulsar and built-in connector

Before connecting Pulsar to a database, you need to install Pulsar and the desired built-in connector.

Read [Run a standalone Pulsar cluster locally](getting-started-standalone.md) for downloading the Pulsar distribution.

To enable Pulsar connectors, you need to download the connectors' tarball release on [download page](pathname:///download/).

After you download the NAR file, copy the file to the `connectors` directory in the Pulsar directory. For example, if you download the `pulsar-io-aerospike-@pulsar:version@.nar` connector file, enter the following commands:

```bash
mkdir connectors
mv pulsar-io-aerospike-@pulsar:version@.nar connectors

ls connectors
# pulsar-io-aerospike-@pulsar:version@.nar
# ...
```

:::note

* If you are running Pulsar in a bare metal cluster, make sure `connectors` tarball is unzipped in every pulsar directory of the broker (or in every pulsar directory of function-worker if you are running a separate worker cluster for Pulsar Functions).
* If you are [running Pulsar in Docker](getting-started-docker.md) or deploying Pulsar using a docker image (e.g. [K8S](deploy-kubernetes.md)), you can use the `apachepulsar/pulsar-all` image instead of the `apachepulsar/pulsar` image. The `apachepulsar/pulsar-all` image has already bundled all built-in connectors.

:::

## Start Pulsar standalone

1. Start Pulsar locally.

   ```bash
   bin/pulsar standalone
   ```

   All the components of a Pulsar service are started in order.

   You can curl those pulsar service endpoints to make sure the Pulsar service is up and running correctly.

2. Check Pulsar binary protocol port.

   ```bash
   telnet localhost 6650
   ```

3. Check Pulsar Function cluster.

   ```bash
   curl -s http://localhost:8080/admin/v2/worker/cluster
   ```

   **Example output**

   ```json
   [{"workerId":"c-standalone-fw-localhost-6750","workerHostname":"localhost","port":6750}]
   ```

4. Make sure a public tenant and a default namespace exist.

   ```bash
   curl -s http://localhost:8080/admin/v2/namespaces/public
   ```

   **Example output**

   ```json
   ["public/default","public/functions"]
   ```

5. All built-in connectors should be listed as available.

   ```bash
   curl -s http://localhost:8080/admin/v2/functions/connectors
   ```

   **Example output**

   ```json
   [{"name":"aerospike","description":"Aerospike database sink","sinkClass":"org.apache.pulsar.io.aerospike.AerospikeStringSink"},{"name":"cassandra","description":"Writes data into Cassandra","sinkClass":"org.apache.pulsar.io.cassandra.CassandraStringSink"},{"name":"kafka","description":"Kafka source and sink connector","sourceClass":"org.apache.pulsar.io.kafka.KafkaStringSource","sinkClass":"org.apache.pulsar.io.kafka.KafkaBytesSink"},{"name":"kinesis","description":"Kinesis sink connector","sinkClass":"org.apache.pulsar.io.kinesis.KinesisSink"},{"name":"rabbitmq","description":"RabbitMQ source connector","sourceClass":"org.apache.pulsar.io.rabbitmq.RabbitMQSource"}]
   ```

   If an error occurs when starting the Pulsar service, you may see an exception at the terminal running `pulsar/standalone`,
   or you can navigate to the `logs` directory under the Pulsar directory to view the logs.

## Connect Pulsar to Cassandra and ScyllaDB

This section demonstrates how to connect Pulsar to Cassandra and ScyllaDB.

:::tip

* Make sure you have Docker installed. If you do not have one, see [install Docker](https://docs.docker.com/docker-for-mac/install/). For more information about Docker commands, see [Docker CLI](https://docs.docker.com/engine/reference/commandline/run/).
* The Cassandra sink connector reads messages from Pulsar topics and writes the messages into Cassandra or ScyllaDB tables. For more information, see [Cassandra sink connector](io-cassandra-sink.md).
* **ScyllaDB compatibility**: [ScyllaDB](https://www.scylladb.com/) is a drop-in replacement for Cassandra that maintains full CQL protocol compatibility. The same `pulsar-io-cassandra` connector works with both databases without any modifications. For more information, see [Streaming Real-Time Chat Messages into ScyllaDB with Apache Pulsar](https://www.scylladb.com/2022/04/25/streaming-real-time-chat-messages-into-scylladb-with-apache-pulsar/).

:::

To connect Pulsar to Cassandra or ScyllaDB, you can follow the steps below:

### Step 1: Set up a Cassandra or ScyllaDB cluster

This example uses `cassandra` Docker image to start a single-node Cassandra cluster in Docker. Alternatively, you can use ScyllaDB as shown below.

#### Option A: Using Cassandra

1. Start a Cassandra cluster.

   ```bash
   docker run -d --rm --name=cassandra -p 9042:9042 cassandra:3.11
   ```

   :::note

   Before moving to the next steps, make sure the Cassandra cluster is running.

   :::

2. Make sure the Docker process is running.

   ```bash
   docker ps
   ```

3. Check the Cassandra logs to make sure the Cassandra process is running as expected.

   ```bash
   docker logs cassandra
   ```

4. Check the status of the Cassandra cluster.

   ```bash
   docker exec cassandra nodetool status
   ```

   **Example output**

   ```
   Datacenter: datacenter1
   =======================
   Status=Up/Down
   |/ State=Normal/Leaving/Joining/Moving
   --  Address     Load       Tokens       Owns (effective)  Host ID                               Rack
   UN  172.17.0.2  103.67 KiB  256          100.0%            af0e4b2f-84e0-4f0b-bb14-bd5f9070ff26  rack1
   ```

5. Use `cqlsh` to connect to the Cassandra cluster.

   ```bash
   docker exec -ti cassandra cqlsh localhost
   ```

    **Output**
   ```
   Connected to Test Cluster at localhost:9042.
   [cqlsh 5.0.1 | Cassandra 3.11.2 | CQL spec 3.4.4 | Native protocol v4]
   Use HELP for help.
   cqlsh>
   ```

6. Create a keyspace `pulsar_test_keyspace`.

   ```bash
   cqlsh> CREATE KEYSPACE pulsar_test_keyspace WITH replication = {'class':'SimpleStrategy', 'replication_factor':1};
   ```

7. Create a table `pulsar_test_table`.

   ```bash
   cqlsh> USE pulsar_test_keyspace;
   cqlsh:pulsar_test_keyspace> CREATE TABLE pulsar_test_table (key text PRIMARY KEY, col text);
   ```

#### Option B: Using ScyllaDB

1. Start a ScyllaDB cluster.

   ```bash
   docker run -d --rm --name=scylladb -p 9042:9042 \
       scylladb/scylla:latest \
       --smp 1 --memory 750M --overprovisioned 1
   ```

   :::note

   ScyllaDB requires specific flags when running in containers. The flags above configure it for single-core operation with limited resources.

   :::

2. Make sure the Docker process is running.

   ```bash
   docker ps
   ```

3. Check the ScyllaDB logs to make sure the ScyllaDB process is running as expected.

   ```bash
   docker logs scylladb
   ```

   Wait for ScyllaDB to be ready (this may take 30-60 seconds). Look for messages indicating the database is ready to accept CQL connections.

4. Check the status of the ScyllaDB cluster.

   ```bash
   docker exec scylladb nodetool status
   ```

5. Use `cqlsh` to connect to the ScyllaDB cluster.

   ```bash
   docker exec -ti scylladb cqlsh
   ```

6. Create a keyspace `pulsar_test_keyspace`.

   ```bash
   cqlsh> CREATE KEYSPACE pulsar_test_keyspace WITH replication = {'class':'SimpleStrategy', 'replication_factor':1};
   ```

7. Create a table `pulsar_test_table`.

   ```bash
   cqlsh> USE pulsar_test_keyspace;
   cqlsh:pulsar_test_keyspace> CREATE TABLE pulsar_test_table (key text PRIMARY KEY, col text);
   ```

### Step 2: Configure a Cassandra sink

Now that we have a Cassandra or ScyllaDB cluster running locally.

In this section, you need to configure a Cassandra sink connector. The same connector works for both Cassandra and ScyllaDB.

To run a Cassandra sink connector, you need to prepare a configuration file including the information that Pulsar connector runtime needs to know.

For example, how Pulsar connector can find the Cassandra cluster, what is the keyspace and the table that Pulsar connector uses for writing Pulsar messages to, and so on.

You can create a configuration file through one of the following methods.

* JSON

  ```json
  {
      "roots": "localhost:9042",
      "keyspace": "pulsar_test_keyspace",
      "columnFamily": "pulsar_test_table",
      "keyname": "key",
      "columnName": "col"
  }
  ```

* YAML

  ```yaml
  configs:
      roots: "localhost:9042"
      keyspace: "pulsar_test_keyspace"
      columnFamily: "pulsar_test_table"
      keyname: "key"
      columnName: "col"
  ```

:::note

For ScyllaDB, the configuration is identical. If you used a different container name or hostname, update the `roots` parameter accordingly (e.g., `"scylladb:9042"` if connecting from within a Docker network).

:::

For more information, see [Cassandra sink connector](io-cassandra-sink.md).

### Step 3: Create a Cassandra sink

You can use the [Connector Admin CLI](/reference/#/@pulsar:version_reference@/pulsar-admin/) to create a sink connector and perform other operations on them.

Run the following command to create a Cassandra sink connector with sink type _cassandra_ and the config file _examples/cassandra-sink.yml_ created previously.

:::note

The `sink-type` parameter of the currently built-in connectors is determined by the setting of the `name` parameter specified in the _pulsar-io.yaml_ file.

:::

```bash
bin/pulsar-admin sinks create \
    --tenant public \
    --namespace default \
    --name cassandra-test-sink \
    --sink-type cassandra \
    --sink-config-file $PWD/examples/cassandra-sink.yml \
    --inputs test_cassandra
```

Once the command is executed, Pulsar creates the sink connector _cassandra-test-sink_.

This sink connector runs as a Pulsar Function and writes the messages produced in the topic _test_cassandra_ to the Cassandra table _pulsar_test_table_.

### Step 4: Inspect a Cassandra sink

You can use the [Connector Admin CLI](/reference/#/@pulsar:version_reference@/pulsar-admin/) to monitor a connector and perform other operations on it.

* Get the information of a Cassandra sink.

  ```bash
  bin/pulsar-admin sinks get \
    --tenant public \
    --namespace default \
    --name cassandra-test-sink
  ```

  **Example output**

  ```json
  {
  "tenant": "public",
  "namespace": "default",
  "name": "cassandra-test-sink",
  "className": "org.apache.pulsar.io.cassandra.CassandraStringSink",
  "inputSpecs": {
    "test_cassandra": {
      "isRegexPattern": false
    }
  },
  "configs": {
    "roots": "localhost:9042",
    "keyspace": "pulsar_test_keyspace",
    "columnFamily": "pulsar_test_table",
    "keyname": "key",
    "columnName": "col"
  },
  "parallelism": 1,
  "processingGuarantees": "ATLEAST_ONCE",
  "retainOrdering": false,
  "autoAck": true,
  "archive": "builtin://cassandra"
  }
  ```

* Check the status of a Cassandra sink.

  ```bash
  bin/pulsar-admin sinks status \
    --tenant public \
    --namespace default \
    --name cassandra-test-sink
  ```

  **Example output**

  ```json
  {
  "numInstances" : 1,
  "numRunning" : 1,
  "instances" : [ {
    "instanceId" : 0,
    "status" : {
      "running" : true,
      "error" : "",
      "numRestarts" : 0,
      "numReadFromPulsar" : 0,
      "numSystemExceptions" : 0,
      "latestSystemExceptions" : [ ],
      "numSinkExceptions" : 0,
      "latestSinkExceptions" : [ ],
      "numWrittenToSink" : 0,
      "lastReceivedTime" : 0,
      "workerId" : "c-standalone-fw-localhost-8080"
    }
  } ]
  }
  ```

### Step 5: Verify a Cassandra sink

1. Produce some messages to the input topic of the Cassandra sink _test_cassandra_.

   ```bash
   for i in {0..9}; do bin/pulsar-client produce -m "key-$i" -n 1 test_cassandra; done
   ```

2. Inspect the status of the Cassandra sink _test_cassandra_.

   ```bash
   bin/pulsar-admin sinks status \
       --tenant public \
       --namespace default \
       --name cassandra-test-sink
   ```

   You can see 10 messages are processed by the Cassandra sink _test_cassandra_.

   **Example output**

   ```json
   {
     "numInstances" : 1,
     "numRunning" : 1,
     "instances" : [ {
       "instanceId" : 0,
       "status" : {
         "running" : true,
         "error" : "",
         "numRestarts" : 0,
         "numReadFromPulsar" : 10,
         "numSystemExceptions" : 0,
         "latestSystemExceptions" : [ ],
         "numSinkExceptions" : 0,
         "latestSinkExceptions" : [ ],
         "numWrittenToSink" : 10,
         "lastReceivedTime" : 1551685489136,
         "workerId" : "c-standalone-fw-localhost-8080"
       }
     } ]
   }
   ```

3. Use `cqlsh` to connect to the Cassandra cluster.

   ```bash
   docker exec -ti cassandra cqlsh localhost
   ```

4. Check the data of the Cassandra table _pulsar_test_table_.

   ```bash
   cqlsh> use pulsar_test_keyspace;
   cqlsh:pulsar_test_keyspace> select * from pulsar_test_table;

   key    | col
   --------+--------
    key-5 |  key-5
    key-0 |  key-0
    key-9 |  key-9
    key-2 |  key-2
    key-1 |  key-1
    key-3 |  key-3
    key-6 |  key-6
    key-7 |  key-7
    key-4 |  key-4
    key-8 |  key-8
   ```

### Step 6: Delete a Cassandra Sink

You can use the [Connector Admin CLI](/reference/#/@pulsar:version_reference@/pulsar-admin/)
to delete a connector and perform other operations on it.

```bash
bin/pulsar-admin sinks delete \
    --tenant public \
    --namespace default \
    --name cassandra-test-sink
```

## Connect Pulsar to PostgreSQL

This section demonstrates how to connect Pulsar to PostgreSQL.

:::tip

* Make sure you have Docker installed. If you do not have one, see [install Docker](https://docs.docker.com/docker-for-mac/install/). For more information about Docker commands, see [Docker CLI](https://docs.docker.com/engine/reference/commandline/run/).
* The JDBC sink connector pulls messages from Pulsar topics and persists the messages to ClickHouse, MariaDB, PostgreSQL, or SQLite. For more information, see [JDBC sink connector](io-jdbc-sink.md).

:::

To connect Pulsar to PostgreSQL, you can follow the steps below:

### Step 1: Set up a PostgreSQL cluster

This example uses the PostgreSQL 12 docker image to start a single-node PostgreSQL cluster in Docker.

1. Pull the PostgreSQL 12 image from Docker.

   ```bash
   docker pull postgres:12
   ```

2. Start PostgreSQL.

    ```bash
    docker run -d -it --rm \
        --name pulsar-postgres \
        -p 5432:5432 \
        -e POSTGRES_PASSWORD=password \
        -e POSTGRES_USER=postgres \
        postgres:12
    ```

3. Check if PostgreSQL has been started successfully.

   ```bash
   docker logs -f pulsar-postgres
   ```

   PostgreSQL has been started successfully if the following message appears.

   ```text
   2020-05-11 20:09:24.492 UTC [1] LOG:  starting PostgreSQL 12.2 (Debian 12.2-2.pgdg100+1) on x86_64-pc-linux-gnu, compiled by gcc (Debian 8.3.0-6) 8.3.0, 64-bit
   2020-05-11 20:09:24.492 UTC [1] LOG:  listening on IPv4 address "0.0.0.0", port 5432
   2020-05-11 20:09:24.492 UTC [1] LOG:  listening on IPv6 address "::", port 5432
   2020-05-11 20:09:24.499 UTC [1] LOG:  listening on Unix socket "/var/run/postgresql/.s.PGSQL.5432"
   2020-05-11 20:09:24.523 UTC [55] LOG:  database system was shut down at 2020-05-11 20:09:24 UTC
   2020-05-11 20:09:24.533 UTC [1] LOG:  database system is ready to accept connections
   ```

4. Access to PostgreSQL container.

   ```bash
   docker exec -it pulsar-postgres /bin/bash
   ```

5. Log in to PostgreSQL with the default username and password:

   ```bash
   psql -U postgres postgres
   ```

6. Create a `pulsar_postgres_jdbc_sink` table using the following command:

   ```sql
   create table if not exists pulsar_postgres_jdbc_sink
   (
   id serial PRIMARY KEY,
   name VARCHAR(255) NOT NULL
   );
   ```

### Step 2: Configure a JDBC sink

Now we have a PostgreSQL running locally.

In this section, you need to configure a JDBC sink connector.

1. Add a configuration file.

   To run a JDBC sink connector, you need to prepare a YAML configuration file including the information that the Pulsar connector runtime needs to know.

   For example, how Pulsar connector can find the PostgreSQL cluster, what is the JDBC URL and the table that Pulsar connector uses for writing messages.

   Create a _pulsar-postgres-jdbc-sink.yaml_ file, copy the following contents to this file, and place the file in the `pulsar/connectors` folder.

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

2. Create a schema.

   Create a _avro-schema_ file, copy the following contents to this file, and place the file in the `pulsar/connectors` folder.

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

   :::tip

   For more information about AVRO, see [Apache Avro](https://avro.apache.org/docs/1.9.1/).

   :::

3. Upload a schema to a topic.

   This example uploads the _avro-schema_ schema to the _pulsar-postgres-jdbc-sink-topic_ topic.

   ```bash
   bin/pulsar-admin schemas upload pulsar-postgres-jdbc-sink-topic -f ./connectors/avro-schema
   ```

4. Check if the schema has been uploaded successfully.

   ```bash
   bin/pulsar-admin schemas get pulsar-postgres-jdbc-sink-topic
   ```

   The schema has been uploaded successfully if the following message appears.

   ```json
   {"name":"pulsar-postgres-jdbc-sink-topic","schema":"{\"type\":\"record\",\"name\":\"Test\",\"fields\":[{\"name\":\"id\",\"type\":[\"null\",\"int\"]},{\"name\":\"name\",\"type\":[\"null\",\"string\"]}]}","type":"AVRO","properties":{}}
   ```

### Step 3: Create a JDBC sink

You can use the [Connector Admin CLI](/reference/#/@pulsar:version_reference@/pulsar-admin/)
to create a sink connector and perform other operations on it.

This example creates a sink connector and specifies the desired information.

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

Once the command is executed, Pulsar creates a sink connector _pulsar-postgres-jdbc-sink_.

This sink connector runs as a Pulsar Function and writes the messages produced in the topic _pulsar-postgres-jdbc-sink-topic_ to the PostgreSQL table _pulsar_postgres_jdbc_sink_.

 #### Tip

 Flag | Description | Example
 ---|---|---|
 `--archive` | The absolute path to the archive file for the sink. | $PWD/pulsar-io-jdbc-postgres-@pulsar:version@.nar |
 `--inputs` | The input topic(s) of the sink. <br /><br /> Multiple topics can be specified as a comma-separated list.||
 `--name` | The name of the sink. | pulsar-postgres-jdbc-sink |
 `--sink-config-file` | The absolute path to a YAML config file specifying the configuration of the sink. | $PWD/pulsar-postgres-jdbc-sink.yaml |
 `--parallelism` | The parallelism factor of the sink. <br /><br /> For example, the number of sink instances to run. |  1 |

:::tip

For more information about `pulsar-admin sinks create options`, see [Pulsar admin docs](/reference/#/@pulsar:version_reference@/pulsar-admin/).

:::

The sink has been created successfully if the following message appears.

```bash
Created successfully
```

### Step 4: Inspect a JDBC sink

You can use the [Connector Admin CLI](/reference/#/@pulsar:version_reference@/pulsar-admin/)
to monitor a connector and perform other operations on it.

* List all running JDBC sink(s).

    ```bash
    bin/pulsar-admin sinks list \
        --tenant public \
        --namespace default
    ```

  :::tip

  For more information about `pulsar-admin sinks list options`, see [Pulsar admin docs](/reference/#/@pulsar:version_reference@/pulsar-admin/).

  :::

  The result shows that only the _postgres-jdbc-sink_ sink is running.

  ```json
  [
  "pulsar-postgres-jdbc-sink"
  ]
  ```

* Get the information of a JDBC sink.

    ```bash
    bin/pulsar-admin sinks get \
      --tenant public \
      --namespace default \
      --name pulsar-postgres-jdbc-sink
    ```

  :::tip

  For more information about `pulsar-admin sinks get options`, see [Pulsar admin docs](/reference/#/@pulsar:version_reference@/pulsar-admin/).

  :::

  The result shows the information of the sink connector, including tenant, namespace, topic and so on.

  ```json
  {
  "tenant": "public",
  "namespace": "default",
  "name": "pulsar-postgres-jdbc-sink",
  "className": "org.apache.pulsar.io.jdbc.PostgresJdbcAutoSchemaSink",
  "inputSpecs": {
    "pulsar-postgres-jdbc-sink-topic": {
      "isRegexPattern": false
    }
  },
  "configs": {
    "password": "password",
    "jdbcUrl": "jdbc:postgresql://localhost:5432/pulsar_postgres_jdbc_sink",
    "userName": "postgres",
    "tableName": "pulsar_postgres_jdbc_sink"
  },
  "parallelism": 1,
  "processingGuarantees": "ATLEAST_ONCE",
  "retainOrdering": false,
  "autoAck": true
  }
  ```

* Get the status of a JDBC sink

  ```bash
  bin/pulsar-admin sinks status \
    --tenant public \
    --namespace default \
    --name pulsar-postgres-jdbc-sink
  ```

  :::tip

  For more information about `pulsar-admin sinks status options`, see [Pulsar admin docs](/reference/#/@pulsar:version_reference@/pulsar-admin/).

  :::

  The result shows the current status of the sink connector, including the number of instances, running status, worker ID and so on.

  ```json
  {
  "numInstances" : 1,
  "numRunning" : 1,
  "instances" : [ {
    "instanceId" : 0,
    "status" : {
      "running" : true,
      "error" : "",
      "numRestarts" : 0,
      "numReadFromPulsar" : 0,
      "numSystemExceptions" : 0,
      "latestSystemExceptions" : [ ],
      "numSinkExceptions" : 0,
      "latestSinkExceptions" : [ ],
      "numWrittenToSink" : 0,
      "lastReceivedTime" : 0,
      "workerId" : "c-standalone-fw-192.168.2.52-8080"
    }
  } ]
  }
  ```

### Step 5: Stop a JDBC sink

You can use the [Connector Admin CLI](/reference/#/@pulsar:version_reference@/pulsar-admin/) to stop a connector and perform other operations on it.

```bash
bin/pulsar-admin sinks stop \
    --tenant public \
    --namespace default \
    --name pulsar-postgres-jdbc-sink
```

:::tip

For more information about `pulsar-admin sinks stop options`, see [Pulsar admin docs](/reference/#/@pulsar:version_reference@/pulsar-admin/).

:::

The sink instance has been stopped successfully if the following message disappears.

```bash
Stopped successfully
```

### Step 6: Restart a JDBC sink

You can use the [Connector Admin CLI](/reference/#/@pulsar:version_reference@/pulsar-admin/) to restart a connector and perform other operations on it.

```bash
bin/pulsar-admin sinks restart \
    --tenant public \
    --namespace default \
    --name pulsar-postgres-jdbc-sink
```

:::tip

For more information about `pulsar-admin sinks restart options`, see [Pulsar admin docs](/reference/#/@pulsar:version_reference@/pulsar-admin/).

:::

The sink instance has been started successfully if the following message disappears.

```bash
Started successfully
```

:::tip

* Optionally, you can run a standalone sink connector using `pulsar-admin sinks localrun options`.
Note that `pulsar-admin sinks localrun options` **runs a sink connector locally**, while `pulsar-admin sinks start options` **starts a sink connector in a cluster**.
* For more information about `pulsar-admin sinks localrun options`, see [Pulsar admin docs](/reference/#/@pulsar:version_reference@/pulsar-admin/).

:::

### Step 7: Update a JDBC sink

You can use the [Connector Admin CLI](/reference/#/@pulsar:version_reference@/pulsar-admin/) to update a connector and perform other operations on it.

This example updates the parallelism of the _pulsar-postgres-jdbc-sink_ sink connector to 2.

```bash
bin/pulsar-admin sinks update \
    --name pulsar-postgres-jdbc-sink \
    --parallelism 2
```

:::tip

For more information about `pulsar-admin sinks update options`, see [Pulsar admin docs](/reference/#/@pulsar:version_reference@/pulsar-admin/).

:::

The sink connector has been updated successfully if the following message disappears.

```bash
Updated successfully
```

This example double-checks the information.

```bash
bin/pulsar-admin sinks get \
    --tenant public \
    --namespace default \
    --name pulsar-postgres-jdbc-sink
```

The result shows that the parallelism is 2.

```json
{
  "tenant": "public",
  "namespace": "default",
  "name": "pulsar-postgres-jdbc-sink",
  "className": "org.apache.pulsar.io.jdbc.PostgresJdbcAutoSchemaSink",
  "inputSpecs": {
    "pulsar-postgres-jdbc-sink-topic": {
      "isRegexPattern": false
    }
  },
  "configs": {
    "password": "password",
    "jdbcUrl": "jdbc:postgresql://localhost:5432/pulsar_postgres_jdbc_sink",
    "userName": "postgres",
    "tableName": "pulsar_postgres_jdbc_sink"
  },
  "parallelism": 2,
  "processingGuarantees": "ATLEAST_ONCE",
  "retainOrdering": false,
  "autoAck": true
}
```

### Step 8: Delete a JDBC sink

You can use the [Connector Admin CLI](/reference/#/@pulsar:version_reference@/pulsar-admin/)
to delete a connector and perform other operations on it.

This example deletes the _pulsar-postgres-jdbc-sink_ sink connector.

```bash
bin/pulsar-admin sinks delete \
    --tenant public \
    --namespace default \
    --name pulsar-postgres-jdbc-sink
```

:::tip

For more information about `pulsar-admin sinks delete options`, see [Pulsar admin docs](/reference/#/@pulsar:version_reference@/pulsar-admin/).

:::

The sink connector has been deleted successfully if the following message appears.

```text
Deleted successfully
```

This example double-checks the status of the sink connector.

```bash
bin/pulsar-admin sinks get \
    --tenant public \
    --namespace default \
    --name pulsar-postgres-jdbc-sink
```

The result shows that the sink connector does not exist.

```text
HTTP 404 Not Found
Reason: Sink pulsar-postgres-jdbc-sink doesn't exist
```