Testing Airbyte with SingleStoreDB and Apache Pulsar
Tags: singlestoredb, airbyte, apachepulsar
Published: 2022-10-29 01:43:32

## Abstract
During the 1970s, the three-step process to Extract, Transform, and Load (ETL) became popular when working with data warehouses. As processing power has significantly improved and cloud-based systems have become more powerful and cost-effective, Extract, Load and Transform (ELT) has become possible. This article will demonstrate using SingleStoreDB with one popular open-source ELT tool, Airbyte.

## Introduction
Airbyte provides source and destination connectors that enable the construction of pipelines between different systems. In this article, we'll use one of Airbyte's source connectors with SingleStoreDB Cloud and send the data to Apache Pulsar using a destination connector.

## Create a SingleStoreDB Cloud account
A [previous article](https://dev.to/veryfatboy/quick-tip-using-dbt-with-singlestoredb-161g) showed the steps required to create a free SingleStoreDB Cloud account. We'll use **Airbyte Demo Group** as our Workspace Group Name and **airbyte-demo** as our Workspace Name. We'll make a note of our **password** and **host** name.

## Create the database and table
We'll use a subset of the inventory system example from a [previous article](https://medium.com/@VeryFatBoy/using-singlestore-as-a-json-document-database-bc386fc5c563), as it provides a combination of both Relational and JSON data. In the **SQL Editor** in SingleStoreDB Cloud, we'll create a database and a table:

```sql
CREATE DATABASE e_store;

USE e_store;

CREATE TABLE products (
     id INT UNSIGNED NOT NULL AUTO_INCREMENT,
     name VARCHAR(250) NOT NULL,
     brand_id INT UNSIGNED NOT NULL,
     category_id INT UNSIGNED NOT NULL,
     attributes JSON NOT NULL,
     PRIMARY KEY(id),
     INDEX CATEGORY_ID(category_id ASC),
     INDEX BRAND_ID(brand_id ASC)
);
```

## Populate the database table
Let's now populate the **products** table:

```sql
-- Televisions
INSERT INTO products (name, brand_id, category_id, attributes) VALUES
('Prime', '1', '1', '{"screen" : "50 inch", "resolution" : "2048 x 1152 pixels", "ports" : {"hdmi" : 1, "usb" : 3}, "speakers" : {"left" : "10 watt", "right" : "10 watt"}}'),
('Octoview', '1', '1', '{"screen" : "40 inch", "resolution" : "1920 x 1080 pixels", "ports" : {"hdmi" : 1, "usb" : 2}, "speakers" : {"left" : "10 watt", "right" : "10 watt"}}'),
('Dreamer', '1', '1', '{"screen" : "30 inch", "resolution" : "1600 x 900 pixels", "ports" : {"hdmi" : 1, "usb" : 1}, "speakers" : {"left" : "10 watt", "right" : "10 watt"}}'),
('Bravia', '1', '1', '{"screen" : "25 inch", "resolution" : "1366 x 768 pixels", "ports" : {"hdmi" : 1, "usb" : 0}, "speakers" : {"left" : "5 watt", "right" : "5 watt"}}'),
('Proton', '1', '1', '{"screen" : "20 inch", "resolution" : "1280 x 720 pixels", "ports" : {"hdmi" : 0, "usb" : 0}, "speakers" : {"left" : "5 watt", "right" : "5 watt"}}');

-- Mobile Phones
INSERT INTO products (name, brand_id, category_id, attributes) VALUES
('Desire', '2', '2', JSON_BUILD_OBJECT("network",
        JSON_ARRAY_PUSH_STRING('["GSM", "CDMA", "HSPA"]', 'EVDO'),
        "body",
        "5.11 x 2.59 x 0.46 inches",
        "weight",
        "143 grams",
        "sim",
        "Micro-SIM",
        "display",
        "4.5 inches",
        "resolution",
        "720 x 1280 pixels",
        "os",
        "Android Jellybean v4.3"
    )
),
('Passion', '2', '2', JSON_BUILD_OBJECT("network",
        JSON_ARRAY_PUSH_STRING('["GSM", "CDMA"]', 'HSPA'),
        "body",
        "6.11 x 3.59 x 0.46 inches",
        "weight",
        "145 grams",
        "sim",
        "Micro-SIM",
        "display",
        "4.5 inches",
        "resolution",
        "720 x 1280 pixels",
        "os",
        "Android Jellybean v4.3"
    )
),
('Emotion', '2', '2', JSON_BUILD_OBJECT("network" ,
        JSON_ARRAY_PUSH_STRING('["GSM", "CDMA"]', 'EVDO'),
        "body",
        "5.50 x 2.50 x 0.50 inches",
        "weight",
        "125 grams",
        "sim",
        "Micro-SIM",
        "display",
        "5.00 inches",
        "resolution",
        "720 x 1280 pixels",
        "os",
        "Android KitKat v4.3"
    )
),
('Sensation', '2', '2', JSON_BUILD_OBJECT("network",
        JSON_ARRAY_PUSH_STRING('["GSM", "HSPA"]', 'EVDO'),
        "body",
        "4.00 x 2.00 x 0.75 inches",
        "weight",
        "150 grams",
        "sim",
        "Micro-SIM",
        "display",
        "3.5 inches",
        "resolution",
        "720 x 1280 pixels",
        "os",
        "Android Lollipop v4.3"
    )
),
('Joy', '2', '2', JSON_BUILD_OBJECT("network",
        JSON_ARRAY_PUSH_STRING('["CDMA", "HSPA"]', 'EVDO'),
        "body",
        "7.00 x 3.50 x 0.25 inches",
        "weight",
        "250 grams",
        "sim",
        "Micro-SIM",
        "display",
        "6.5 inches",
        "resolution",
        "1920 x 1080 pixels",
        "os",
        "Android Marshmallow v4.3"
    )
);

-- Cameras
INSERT INTO products (name, brand_id, category_id, attributes) VALUES
('Explorer', '3', '3', '{"sensor_type" : "CMOS", "processor" : "Digic DV III", "scanning_system" : "progressive", "mount_type" : "PL", "monitor_type" : "LCD"}'),
('Runner', '3', '3',   '{"sensor_type" : "CMOS", "processor" : "Digic DV II",  "scanning_system" : "progressive", "mount_type" : "PL", "monitor_type" : "LED"}'),
('Traveler', '3', '3', '{"sensor_type" : "CMOS", "processor" : "Digic DV II",  "scanning_system" : "progressive", "mount_type" : "PL", "monitor_type" : "LCD"}'),
('Walker', '3', '3',   '{"sensor_type" : "CMOS", "processor" : "Digic DV I",   "scanning_system" : "progressive", "mount_type" : "PL", "monitor_type" : "LED"}'),
('Jumper', '3', '3',   '{"sensor_type" : "CMOS", "processor" : "Digic DV I",   "scanning_system" : "progressive", "mount_type" : "PL", "monitor_type" : "LCD"}');
```

## Install and configure Apache Pulsar
We'll install Pulsar using the [instructions](https://pulsar.apache.org/docs/getting-started-standalone/) on the website:

```shell
wget https://archive.apache.org/dist/pulsar/pulsar-2.10.2/apache-pulsar-2.10.2-bin.tar.gz

tar xvfz apache-pulsar-2.10.2-bin.tar.gz

cd apache-pulsar-2.10.2
```

Next, we'll start Pulsar standalone:

```shell
bin/pulsar standalone
```

From another terminal window, we'll run a `consumer`, as follows:

```shell
bin/pulsar-client consume sample_topic -s "first-subscription" -n 0
```

The consumer will now wait for messages. We'll send messages to `sample_topic` from Airbyte.

## Install and configure Airbyte
We'll install Airbyte open-source using the [instructions](https://docs.airbyte.com/quickstart/deploy-airbyte/) on the website:

```shell
git clone https://github.com/airbytehq/airbyte.git

cd airbyte

docker-compose up
```

After the deployment is complete, we'll launch a web browser and enter `http://localhost:8000`.

We'll be asked for a username and password. The defaults are `airbyte` and `password`, respectively.

### Create Source
Once logged in, we'll select `Sources` from the left-hand navigation pane and create a source.

We'll choose MySQL as the **Source type** on the **New connection page**.

On the **Set up the source** page, here is what we need to enter:

- **Source type:** MySQL
- **Source name:** SingleStoreDB
- **Host:** &lt;host&gt;
- **Port:** 3306
- **Database:** e_store
- **Username:** admin
- **Password:** &lt;password&gt;
- **SSL Connection:** Option is enabled

We'll replace the `<host>` and `<password>` with the values from our SingleStoreDB Cloud account.

Next, we'll click the **Set up source** button. The connection should be successful.

### Create Destination
We'll choose Pulsar as the **Destination type** on the **New connection page**.

On the **Set up the destination** page, here is what we need to enter:

- **Destination type:** Pulsar
- **Destination name:** Pulsar
- **Pulsar brokers:** localhost:6650
- **Test topic:** test_topic
- **Topic type:** persistent
- **Topic tenant:** public
- **Producer name:** airbyte_producer
- **Topic pattern:** sample_topic

Next, we'll click the **Set up destination** button. The connection should be successful.

### Create Connection
We'll now link our source to our destination.

We'll select SingleStoreDB and click the **Use existing source** button.

We'll select Pulsar and click the **Use existing destination** button.

For our initial test, we'll configure the connection as follows:

- **Connection name:** SingleStoreDB &lt;&gt; Pulsar
- **Transfer &gt; Replication frequency:** Every 24 hours
- **Streams &gt; Destination Namespace:** Mirror source structure
- **Activate the streams you want to sync:** Check (✔) the products table and select **Full refresh | Append** for sync mode

Next, we'll click the **Set up** connection button. Sync should start immediately.

In the terminal window where our consumer is running, we should see output similar to the following:

```
----- got message -----
key:[null], properties:[], content:{"_airbyte_ab_id":"10a95b94-ca07-48c3-8655-04e1160684ca","_airbyte_data":"eyJpZCI6MiwibmFtZSI6Ik9jdG92aWV3IiwiYnJhbmRfaWQiOjEsImNhdGVnb3J5X2lkIjoxLCJhdHRyaWJ1dGVzIjoie1wicG9ydHNcIjp7XCJoZG1pXCI6MSxcInVzYlwiOjJ9LFwicmVzb2x1dGlvblwiOlwiMTkyMCB4IDEwODAgcGl4ZWxzXCIsXCJzY3JlZW5cIjpcIjQwIGluY2hcIixcInNwZWFrZXJzXCI6e1wibGVmdFwiOlwiMTAgd2F0dFwiLFwicmlnaHRcIjpcIjEwIHdhdHRcIn19In0=","_airbyte_stream":"products","_airbyte_emitted_at":1666989683549}
----- got message -----
key:[null], properties:[], content:{"_airbyte_ab_id":"ac9a6d74-192f-4133-a77b-b2783a5a86a6","_airbyte_data":"eyJpZCI6NCwibmFtZSI6IkJyYXZpYSIsImJyYW5kX2lkIjoxLCJjYXRlZ29yeV9pZCI6MSwiYXR0cmlidXRlcyI6IntcInBvcnRzXCI6e1wiaGRtaVwiOjEsXCJ1c2JcIjowfSxcInJlc29sdXRpb25cIjpcIjEzNjYgeCA3NjggcGl4ZWxzXCIsXCJzY3JlZW5cIjpcIjI1IGluY2hcIixcInNwZWFrZXJzXCI6e1wibGVmdFwiOlwiNSB3YXR0XCIsXCJyaWdodFwiOlwiNSB3YXR0XCJ9fSJ9","_airbyte_stream":"products","_airbyte_emitted_at":1666989683549}
----- got message -----
key:[null], properties:[], content:{"_airbyte_ab_id":"51a5a79f-97a0-47e5-9a9c-55b36483d6c4","_airbyte_data":"eyJpZCI6NSwibmFtZSI6IlByb3RvbiIsImJyYW5kX2lkIjoxLCJjYXRlZ29yeV9pZCI6MSwiYXR0cmlidXRlcyI6IntcInBvcnRzXCI6e1wiaGRtaVwiOjAsXCJ1c2JcIjowfSxcInJlc29sdXRpb25cIjpcIjEyODAgeCA3MjAgcGl4ZWxzXCIsXCJzY3JlZW5cIjpcIjIwIGluY2hcIixcInNwZWFrZXJzXCI6e1wibGVmdFwiOlwiNSB3YXR0XCIsXCJyaWdodFwiOlwiNSB3YXR0XCJ9fSJ9","_airbyte_stream":"products","_airbyte_emitted_at":1666989683549}
----- got message -----
key:[null], properties:[], content:{"_airbyte_ab_id":"745fc969-3d29-4866-973b-db7fde3ad9f4","_airbyte_data":"eyJpZCI6NywibmFtZSI6IlBhc3Npb24iLCJicmFuZF9pZCI6MiwiY2F0ZWdvcnlfaWQiOjIsImF0dHJpYnV0ZXMiOiJ7XCJib2R5XCI6XCI2LjExIHggMy41OSB4IDAuNDYgaW5jaGVzXCIsXCJkaXNwbGF5XCI6XCI0LjUgaW5jaGVzXCIsXCJuZXR3b3JrXCI6W1wiR1NNXCIsXCJDRE1BXCIsXCJIU1BBXCJdLFwib3NcIjpcIkFuZHJvaWQgSmVsbHliZWFuIHY0LjNcIixcInJlc29sdXRpb25cIjpcIjcyMCB4IDEyODAgcGl4ZWxzXCIsXCJzaW1cIjpcIk1pY3JvLVNJTVwiLFwid2VpZ2h0XCI6XCIxNDUgZ3JhbXNcIn0ifQ==","_airbyte_stream":"products","_airbyte_emitted_at":1666989683549}
----- got message -----
key:[null], properties:[], content:{"_airbyte_ab_id":"7c64455b-0c5f-4292-a4bc-6ea29dd209f8","_airbyte_data":"eyJpZCI6OCwibmFtZSI6IkVtb3Rpb24iLCJicmFuZF9pZCI6MiwiY2F0ZWdvcnlfaWQiOjIsImF0dHJpYnV0ZXMiOiJ7XCJib2R5XCI6XCI1LjUwIHggMi41MCB4IDAuNTAgaW5jaGVzXCIsXCJkaXNwbGF5XCI6XCI1LjAwIGluY2hlc1wiLFwibmV0d29ya1wiOltcIkdTTVwiLFwiQ0RNQVwiLFwiRVZET1wiXSxcIm9zXCI6XCJBbmRyb2lkIEtpdEthdCB2NC4zXCIsXCJyZXNvbHV0aW9uXCI6XCI3MjAgeCAxMjgwIHBpeGVsc1wiLFwic2ltXCI6XCJNaWNyby1TSU1cIixcIndlaWdodFwiOlwiMTI1IGdyYW1zXCJ9In0=","_airbyte_stream":"products","_airbyte_emitted_at":1666989683549}
----- got message -----
key:[null], properties:[], content:{"_airbyte_ab_id":"2c3ce9ba-56a3-4cb5-90d9-035b79b2c06f","_airbyte_data":"eyJpZCI6MTAsIm5hbWUiOiJKb3kiLCJicmFuZF9pZCI6MiwiY2F0ZWdvcnlfaWQiOjIsImF0dHJpYnV0ZXMiOiJ7XCJib2R5XCI6XCI3LjAwIHggMy41MCB4IDAuMjUgaW5jaGVzXCIsXCJkaXNwbGF5XCI6XCI2LjUgaW5jaGVzXCIsXCJuZXR3b3JrXCI6W1wiQ0RNQVwiLFwiSFNQQVwiLFwiRVZET1wiXSxcIm9zXCI6XCJBbmRyb2lkIE1hcnNobWFsbG93IHY0LjNcIixcInJlc29sdXRpb25cIjpcIjE5MjAgeCAxMDgwIHBpeGVsc1wiLFwic2ltXCI6XCJNaWNyby1TSU1cIixcIndlaWdodFwiOlwiMjUwIGdyYW1zXCJ9In0=","_airbyte_stream":"products","_airbyte_emitted_at":1666989683549}
----- got message -----
key:[null], properties:[], content:{"_airbyte_ab_id":"1e20dd1d-a390-4f3f-96ff-dacc70f4ae41","_airbyte_data":"eyJpZCI6MTIsIm5hbWUiOiJSdW5uZXIiLCJicmFuZF9pZCI6MywiY2F0ZWdvcnlfaWQiOjMsImF0dHJpYnV0ZXMiOiJ7XCJtb25pdG9yX3R5cGVcIjpcIkxFRFwiLFwibW91bnRfdHlwZVwiOlwiUExcIixcInByb2Nlc3NvclwiOlwiRGlnaWMgRFYgSUlcIixcInNjYW5uaW5nX3N5c3RlbVwiOlwicHJvZ3Jlc3NpdmVcIixcInNlbnNvcl90eXBlXCI6XCJDTU9TXCJ9In0=","_airbyte_stream":"products","_airbyte_emitted_at":1666989683549}
----- got message -----
key:[null], properties:[], content:{"_airbyte_ab_id":"b225a1f3-c31f-4ee8-9dd9-922b36d7be96","_airbyte_data":"eyJpZCI6MTUsIm5hbWUiOiJKdW1wZXIiLCJicmFuZF9pZCI6MywiY2F0ZWdvcnlfaWQiOjMsImF0dHJpYnV0ZXMiOiJ7XCJtb25pdG9yX3R5cGVcIjpcIkxDRFwiLFwibW91bnRfdHlwZVwiOlwiUExcIixcInByb2Nlc3NvclwiOlwiRGlnaWMgRFYgSVwiLFwic2Nhbm5pbmdfc3lzdGVtXCI6XCJwcm9ncmVzc2l2ZVwiLFwic2Vuc29yX3R5cGVcIjpcIkNNT1NcIn0ifQ==","_airbyte_stream":"products","_airbyte_emitted_at":1666989683549}
----- got message -----
key:[null], properties:[], content:{"_airbyte_ab_id":"421a202e-6309-473d-b376-16736e6b733d","_airbyte_data":"eyJpZCI6MSwibmFtZSI6IlByaW1lIiwiYnJhbmRfaWQiOjEsImNhdGVnb3J5X2lkIjoxLCJhdHRyaWJ1dGVzIjoie1wicG9ydHNcIjp7XCJoZG1pXCI6MSxcInVzYlwiOjN9LFwicmVzb2x1dGlvblwiOlwiMjA0OCB4IDExNTIgcGl4ZWxzXCIsXCJzY3JlZW5cIjpcIjUwIGluY2hcIixcInNwZWFrZXJzXCI6e1wibGVmdFwiOlwiMTAgd2F0dFwiLFwicmlnaHRcIjpcIjEwIHdhdHRcIn19In0=","_airbyte_stream":"products","_airbyte_emitted_at":1666989683549}
----- got message -----
key:[null], properties:[], content:{"_airbyte_ab_id":"ed03e9bf-907f-4a1f-b051-6615b3ebeb3d","_airbyte_data":"eyJpZCI6MywibmFtZSI6IkRyZWFtZXIiLCJicmFuZF9pZCI6MSwiY2F0ZWdvcnlfaWQiOjEsImF0dHJpYnV0ZXMiOiJ7XCJwb3J0c1wiOntcImhkbWlcIjoxLFwidXNiXCI6MX0sXCJyZXNvbHV0aW9uXCI6XCIxNjAwIHggOTAwIHBpeGVsc1wiLFwic2NyZWVuXCI6XCIzMCBpbmNoXCIsXCJzcGVha2Vyc1wiOntcImxlZnRcIjpcIjEwIHdhdHRcIixcInJpZ2h0XCI6XCIxMCB3YXR0XCJ9fSJ9","_airbyte_stream":"products","_airbyte_emitted_at":1666989683549}
----- got message -----
key:[null], properties:[], content:{"_airbyte_ab_id":"e371792a-79da-4ae1-a644-271c4f0e8b8c","_airbyte_data":"eyJpZCI6NiwibmFtZSI6IkRlc2lyZSIsImJyYW5kX2lkIjoyLCJjYXRlZ29yeV9pZCI6MiwiYXR0cmlidXRlcyI6IntcImJvZHlcIjpcIjUuMTEgeCAyLjU5IHggMC40NiBpbmNoZXNcIixcImRpc3BsYXlcIjpcIjQuNSBpbmNoZXNcIixcIm5ldHdvcmtcIjpbXCJHU01cIixcIkNETUFcIixcIkhTUEFcIixcIkVWRE9cIl0sXCJvc1wiOlwiQW5kcm9pZCBKZWxseWJlYW4gdjQuM1wiLFwicmVzb2x1dGlvblwiOlwiNzIwIHggMTI4MCBwaXhlbHNcIixcInNpbVwiOlwiTWljcm8tU0lNXCIsXCJ3ZWlnaHRcIjpcIjE0MyBncmFtc1wifSJ9","_airbyte_stream":"products","_airbyte_emitted_at":1666989683549}
----- got message -----
key:[null], properties:[], content:{"_airbyte_ab_id":"ffd5ac6c-15cb-4110-9bd0-55125b9528f7","_airbyte_data":"eyJpZCI6OSwibmFtZSI6IlNlbnNhdGlvbiIsImJyYW5kX2lkIjoyLCJjYXRlZ29yeV9pZCI6MiwiYXR0cmlidXRlcyI6IntcImJvZHlcIjpcIjQuMDAgeCAyLjAwIHggMC43NSBpbmNoZXNcIixcImRpc3BsYXlcIjpcIjMuNSBpbmNoZXNcIixcIm5ldHdvcmtcIjpbXCJHU01cIixcIkhTUEFcIixcIkVWRE9cIl0sXCJvc1wiOlwiQW5kcm9pZCBMb2xsaXBvcCB2NC4zXCIsXCJyZXNvbHV0aW9uXCI6XCI3MjAgeCAxMjgwIHBpeGVsc1wiLFwic2ltXCI6XCJNaWNyby1TSU1cIixcIndlaWdodFwiOlwiMTUwIGdyYW1zXCJ9In0=","_airbyte_stream":"products","_airbyte_emitted_at":1666989683549}
----- got message -----
key:[null], properties:[], content:{"_airbyte_ab_id":"769cffc1-6c64-4440-82df-070c45b90990","_airbyte_data":"eyJpZCI6MTEsIm5hbWUiOiJFeHBsb3JlciIsImJyYW5kX2lkIjozLCJjYXRlZ29yeV9pZCI6MywiYXR0cmlidXRlcyI6IntcIm1vbml0b3JfdHlwZVwiOlwiTENEXCIsXCJtb3VudF90eXBlXCI6XCJQTFwiLFwicHJvY2Vzc29yXCI6XCJEaWdpYyBEViBJSUlcIixcInNjYW5uaW5nX3N5c3RlbVwiOlwicHJvZ3Jlc3NpdmVcIixcInNlbnNvcl90eXBlXCI6XCJDTU9TXCJ9In0=","_airbyte_stream":"products","_airbyte_emitted_at":1666989683549}
----- got message -----
key:[null], properties:[], content:{"_airbyte_ab_id":"874f205c-018a-464c-a6fa-8a6dcfafd4d7","_airbyte_data":"eyJpZCI6MTMsIm5hbWUiOiJUcmF2ZWxlciIsImJyYW5kX2lkIjozLCJjYXRlZ29yeV9pZCI6MywiYXR0cmlidXRlcyI6IntcIm1vbml0b3JfdHlwZVwiOlwiTENEXCIsXCJtb3VudF90eXBlXCI6XCJQTFwiLFwicHJvY2Vzc29yXCI6XCJEaWdpYyBEViBJSVwiLFwic2Nhbm5pbmdfc3lzdGVtXCI6XCJwcm9ncmVzc2l2ZVwiLFwic2Vuc29yX3R5cGVcIjpcIkNNT1NcIn0ifQ==","_airbyte_stream":"products","_airbyte_emitted_at":1666989683549}
----- got message -----
key:[null], properties:[], content:{"_airbyte_ab_id":"1929e8b9-c4ac-4ef6-875e-76f44d1640a2","_airbyte_data":"eyJpZCI6MTQsIm5hbWUiOiJXYWxrZXIiLCJicmFuZF9pZCI6MywiY2F0ZWdvcnlfaWQiOjMsImF0dHJpYnV0ZXMiOiJ7XCJtb25pdG9yX3R5cGVcIjpcIkxFRFwiLFwibW91bnRfdHlwZVwiOlwiUExcIixcInByb2Nlc3NvclwiOlwiRGlnaWMgRFYgSVwiLFwic2Nhbm5pbmdfc3lzdGVtXCI6XCJwcm9ncmVzc2l2ZVwiLFwic2Vuc29yX3R5cGVcIjpcIkNNT1NcIn0ifQ==","_airbyte_stream":"products","_airbyte_emitted_at":1666989683549}
2022-10-28T21:41:37,441+0100 [pulsar-timer-6-1] INFO  org.apache.pulsar.client.impl.ConsumerStatsRecorderImpl - [sample_topic] [first-subscription] [ce6e3] Prefetched messages: 0 --- Consume throughput received: 0.25 msgs/s --- 0.00 Mbit/s --- Ack sent rate: 0.25 ack/s --- Failed messages: 0 --- batch messages: 0 ---Failed acks: 0
```

There are 15 messages, one for each row of the products table. The data are encoded into `base64` and if we take the first message above and decode it, we'll get the following:

```json
{"id":2,"name":"Octoview","brand_id":1,"category_id":1,"attributes":"{\"ports\":{\"hdmi\":1,\"usb\":2},\"resolution\":\"1920 x 1080 pixels\",\"screen\":\"40 inch\",\"speakers\":{\"left\":\"10 watt\",\"right\":\"10 watt\"}}"}
```

The data matches one of the rows that we previously stored.

## Summary
In this article, we have seen that we can easily use SingleStoreDB as a source in Airbyte. We successfully replicated the data from a SingleStoreDB Cloud database to Apache Pulsar. Further tests could include other destinations available in Airbyte.