Data Streaming with Kafka in PieCloudDB

Question

I recently started learning Kafka and PieCloudDB database, I want to use Kafka streams to achieve real-time processing. Does PieCloudDB database support loading data from Kafka streams?

How can I configure it to use this feature?

Answer 1

PieCloudDB database supports Kafka streaming data loading.Try this out:

1.Create a topic on Kafka node.

/kafka/kafka_2.13-3.4.0/bin/kafka-topics.sh --create --bootstrap-server (your ip address):9092 --replication-factor 1 --partitions 1 --topic test_tb1

enter image description here

2.Create a table.

psql –p 6003 testdb
drop table test_tb1;
reate table test_tb1(id integer,name varchar(100),age integer);

enter image description here

3.Add extension.

Psql –p 6003 testdb
truncate table test_tb1_ft_offset;
drop EXTENSION kafka_fdw CASCADE;
CREATE EXTENSION kafka_fdw;

enter image description here

4.Add the configuration of PieCloudDB database to the kafka_config.yaml file.

PDB:
  Database:testdb
  User:openpie
  Host:(your host name）
  Port:6003
  SSLMode:disable
Kafka:
  Brokers:(your ip address):9092
  Topic:test_tb
  OffsetTable:ft_offset
  WindowDuration:1000
  Value:input2
  IdlePollInterval: 5000
Input:
  input2:
    Function: jsonb_format 
    Format: jsonb
    Type: jsonbfmt
Output:
  Table: test_tb1
  Columns:
  - Name: id  
    Expression: (input2->>'id')::int
  - Name: name
    Expression: (input2->>'name')::text
  - Name: age
    Expression: (input2->>'age')::int
Foreign:
  Server: mysrv
  Table: ft   
workingSchema: ""

5.Execute the following commands to start Kafka synchronization service.

nohup pdbconduct kafka --debug -c  kafka_config.yaml &
（view the table data）
psql -p 6003 testdb
select count(*) from test_tb1;

6.Kafka produces data.

/kafka/kafka_2.13-3.4.0/bin/kafka-console-producer.sh--bootstrap-server (your ip address):9092 --topic test_tb1
enter：{"id":1,"name":"peter","age":40}
enter again：{"id":2,"name":"jojo","age":30}

enter image description here

7.Query data in PieCloudDB （view the table data again）

psql -p 6003 testdb
select * from test_tb1;

enter image description here