# Upserting into multiple tables from multiples topics using kafka-connect

**URL:** <https://forum.confluent.io/t/upserting-into-multiple-tables-from-multiples-topics-using-kafka-connect/1211>\
**Category:** Kafka Connect\
**Created:** [15 March 2021 21:26 UTC](https://forum.confluent.io/t/upserting-into-multiple-tables-from-multiples-topics-using-kafka-connect/1211 "2021-03-15T21:26:07Z")\
**Posts on this page:** 6\
**Page:** 1

<div class="post-metadata">

**Author:** ![Megna4](https://avatars.discourse-cdn.com/v4/letter/m/8e8cbc/32.png) [@Megna4](https://forum.confluent.io/u/Megna4)\
**Post date:** [15 March 2021 21:26 UTC](https://forum.confluent.io/t/upserting-into-multiple-tables-from-multiples-topics-using-kafka-connect/1211/1 "2021-03-15T21:26:08Z")

</div>

Hello, I’m building a new data pipeline that will move data from Postgres to snowflake with Kafka connect, I will have the Postgres source and a snowflake sink connectors. I want to insert data from around 50 tables from Postgres to 50 Tables on snowflakes using Docker file, Can you please give an Examples of how can I achieve this automatically.

Thanks,  
Megna.

---

<div class="post-metadata">

**Author:** ![rmoff](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/rmoff/32/3_2.png) [@rmoff](https://forum.confluent.io/u/rmoff)\
**Post date:** [16 March 2021 16:00 UTC](https://forum.confluent.io/t/upserting-into-multiple-tables-from-multiples-topics-using-kafka-connect/1211/2 "2021-03-16T16:00:19Z")

</div>

Hi @Megna4,

Depending on which source connector you’re using there’ll be the option to specify a list of tables to include, either by name or a wildcard.

Similarly with the sink connector you should be able to specify `topics.regex` or a comma separated list of `topics` values.

---

<div class="post-metadata">

**Author:** ![Megna4](https://avatars.discourse-cdn.com/v4/letter/m/8e8cbc/32.png) [@Megna4](https://forum.confluent.io/u/Megna4)\
**Post date:** [16 March 2021 16:23 UTC](https://forum.confluent.io/t/upserting-into-multiple-tables-from-multiples-topics-using-kafka-connect/1211/3 "2021-03-16T16:23:38Z")

</div>

Hello Robin, thanks for the Reply. I have around 50 Tables in the Postgres Database. I am guessing I will create a 1 topic to 1 table then how will I be able to pass all the 50 tables in a single Connector in a Curl Command?  
Ex:

```json
curl -vvv -X POST -H "Content-Type: application/json" --data '{
"name":"KafkaSinkConnectortoSnowflakesAccount",
"config":{
		"connector.class":"com.snowflake.kafka.connector.SnowflakeSinkConnector",
		"tasks.max":"8",
		"topics":"sfkafkasftest.salesforce.account",
                "snowflake.topic2table.map": "sfkafkasftest.salesforce.account:ACCOUNT",

```

---

<div class="post-metadata">

**Author:** ![rmoff](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/rmoff/32/3_2.png) [@rmoff](https://forum.confluent.io/u/rmoff)\
**Post date:** [16 March 2021 16:41 UTC](https://forum.confluent.io/t/upserting-into-multiple-tables-from-multiples-topics-using-kafka-connect/1211/4 "2021-03-16T16:41:43Z")

</div>

You should be able to do it all in one connector.

Looking at [the documentation](https://docs.snowflake.com/en/user-guide/kafka-connector-install.html#required-properties) it seems you can use `topics` or `topics.regex` as described above, and then optionally a list of mapping pairs in `snowflake.topic2table.map`.

You could also use the [`RegExRouter` Single Message Transform](https://rmoff.net/2020/12/11/twelve-days-of-smt-day-4-regexrouter/) as part of the connector configuration to rename the topics if they all match a particular wildcard

---

<div class="post-metadata">

**Author:** ![Megna4](https://avatars.discourse-cdn.com/v4/letter/m/8e8cbc/32.png) [@Megna4](https://forum.confluent.io/u/Megna4)\
**Post date:** [22 March 2021 19:56 UTC](https://forum.confluent.io/t/upserting-into-multiple-tables-from-multiples-topics-using-kafka-connect/1211/5 "2021-03-22T19:56:07Z")

</div>

Robin, Thanks for the suggestion on using regex, I am facing a weird Issue and seems like i am unable to find the Information any where.  
When I create a New topic and add a table the records are no flowing but when i make a change on the postgres table after the topic is created like an Insert, Update or delete that’s when the data is flowing. Is there a setting I am missing? I want the Data to flow as soon as the Topic is created.

---

<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:** [21 April 2021 19:56 UTC](https://forum.confluent.io/t/upserting-into-multiple-tables-from-multiples-topics-using-kafka-connect/1211/6 "2021-04-21T19:56:28Z")

</div>

Closed on 21 April 2021 19:56 UTC

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