# How to create a TABLE from STREAM with PRIMARY KEY in ksqldb

**URL:** <https://forum.confluent.io/t/how-to-create-a-table-from-stream-with-primary-key-in-ksqldb/6821>\
**Category:** ksqlDB\
**Created:** [13 December 2022 13:25 UTC](https://forum.confluent.io/t/how-to-create-a-table-from-stream-with-primary-key-in-ksqldb/6821 "2022-12-13T13:25:44Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![leandroluk](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/leandroluk/32/2600_2.png) [@leandroluk](https://forum.confluent.io/u/leandroluk)\
**Post date:** [13 December 2022 13:25 UTC](https://forum.confluent.io/t/how-to-create-a-table-from-stream-with-primary-key-in-ksqldb/6821/1 "2022-12-13T13:25:44Z")

</div>

I’m creating a data conversion pipeline that will have MySQL as SOURCE and ElasticSearch as SYNC. My data source is modeled in MySQL like this:

```sql
CREATE TABLE `person` (
    `_uid` CHAR(36) NOT NULL PRIMARY KEY,
    `_created` DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
    `_updated` DATETIME(3) NULL,
    `_disabled` DATETIME(3) NULL,
    `name` VARCHAR(100) NOT NULL,
    `age` INT NOT NULL
);
CREATE TABLE `contact_type` (
    `_uid` CHAR(36) NOT NULL PRIMARY KEY,
    `value` VARCHAR(100) NOT NULL
);
CREATE TABLE `person_contact` (
    `_uid` CHAR(36) NOT NULL PRIMARY KEY,
    `_created` DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
    `_updated` DATETIME(3) NULL,
    `_disabled` DATETIME(3) NULL,
    `is_default` TINYINT NOT NULL DEFAULT 0,
    `value` TEXT NOT NULL,
    `contact_type_uid` CHAR(36) NOT NULL,
    `person_uid` CHAR(36) NOT NULL,
    FOREIGN KEY (`contact_type_uid`) REFERENCES `contact_type` (`_uid`),
    FOREIGN KEY (`person_uid`) REFERENCES `person` (`_uid`)
);

```

To import the data into ksqldb I created a connector like this:

```sql
CREATE SOURCE CONNECTOR "source-mysql" WITH (
  "connector.class" = 'io.confluent.connect.jdbc.JdbcSourceConnector',
  "connection.url" = 'jdbc:mysql://mysql:3306/db',
  "connection.user" = 'root',
  "connection.password" = 'root',
  "mode" = 'bulk',
  "topic.prefix" = 'source-mysql-'
);

```

And then I created the source’s pointing to the topic’s like this:

```sql
CREATE STREAM "source-mysql-contact_type" (
  "_uid" STRING,
  "value" STRING
) WITH (
  KAFKA_TOPIC='source-mysql-contact_type',
  VALUE_FORMAT='AVRO'
);

```

Next I’m trying to create a table from the STREAM as in the example below:

```sql
CREATE TABLE "table-mysql-contact_type" WITH (
  KAFKA_TOPIC='table-mysql-contact_type',
  VALUE_FORMAT='AVRO'
) AS SELECT
  "_uid",
  LATEST_BY_OFFSET("value") AS "value"
FROM "source-mysql-contact_type"
GROUP BY "_uid"
EMIT CHANGES;

```

The problem is that with this query I can’t define the PRIMARY KEY “\_uid” as it is in MySQL.

I also tried to create the table before and do the insert like this:

```sql
CREATE TABLE "table-mysql-contact_type" (
  "_uid" STRING PRIMARY KEY,
  "value" STRING
) WITH (
  KAFKA_TOPIC='table-mysql-contact_type',
  VALUE_FORMAT='AVRO'
);
INSERT INTO "table-mysql-contact_type"
SELECT
  "_uid",
  LATEST_BY_OFFSET("value") AS "value"
FROM "source-mysql-contact_type"
GROUP BY "_uid"
EMIT CHANGES;

```

But I’m getting the message:

```nohighlight
Exception while preparing statement: INSERT INTO can only be used to insert 
into a stream. table-mysql-contact_type is a table.

```

What do I need to do to be able to create a TABLE that receives data from a STREAM but that I can define the PRIMARY KEY of it?

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/flex019/uploads/confluentcommunity/original/1X/c49438c90c9df282e9996fdf6971be890c71b65a.svg) [@system](https://forum.confluent.io/u/system)\
**Post date:** [12 January 2023 13:26 UTC](https://forum.confluent.io/t/how-to-create-a-table-from-stream-with-primary-key-in-ksqldb/6821/2 "2023-01-12T13:26:34Z")

</div>

This topic was automatically closed after 30 days. New replies are no longer allowed.