인프런 커뮤니티 질문&답변

안녕하십니까 다변 주셔서 감사합니다.

git 으로 제공 해주신 자료를 토대로 제 개발 PC 환경에 마춰 변경되어야 하는 IP,이름 등 이외 모두 복사 하여 사용 하였습니다.

우선 MySQL 에서 Table을 만들었습니다.

KafkaConnect 강의 GIt에서 Debezium Source MySql 에서 테이블 생성 정보로 만들었습니다.

CREATE TABLE customers (
customer_id int NOT NULL PRIMARY KEY,
email_address varchar(255) NOT NULL,
full_name varchar(255) NOT NULL
) ENGINE=InnoDB ;

Source Connect는 환경 설정 부분만 수정 하였습니다. MySql과 연결이 잘되었습니다.

{
    "name": "mysql_cdc_ops_source_avro_03",
    "config": {
        "connector.class": "io.debezium.connector.mysql.MySqlConnector",
        "tasks.max": "1",
        "database.hostname": "192.168.0.26",
        "database.port": "3306",
        "database.user": "cnt_dev",
        "database.password": "1234",
        "database.server.id": "31002",
        "database.server.name": "mysqlavro022",
        "database.include.list": "oc",
        "table.include.list": "oc.customers",
        "database.history.kafka.bootstrap.servers": "localhost:9092",
        "database.history.kafka.topic": "schema-changes.mysql.oc",

        "time.precision.mode": "connect",
        "database.connectionTimezone": "Asia/Seoul",

        "key.converter": "io.confluent.connect.avro.AvroConverter",
        "value.converter": "io.confluent.connect.avro.AvroConverter",
        "key.converter.schema.registry.url": "http://localhost:8081",
        "value.converter.schema.registry.url": "http://localhost:8081",


        "transforms": "unwrap",
        "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
        "transforms.unwrap.drop.tombstones": "false"
    }
}

마지막으로 KsqlDB에 접속 하여 create table을 하면 잘 만들어 집니다.

CREATE TABLE oc_customer (
  customer_id int PRIMARY KEY,  
  email_address varchar,
  full_name varchar  
) WITH (	
	KAFKA_TOPIC = 'mysqlavro022.oc.customers',	
	KEY_FORMAT = 'AVRO',
	VALUE_FORMAT = 'AVRO'
);

이후 조회 명령어를 치면 error가 발생 합니다.

select * from oc_customer emit changes limit 5;
select * from oc_customer emit changes;
select * from oc_customer;

또 의문인 것은 쥬키퍼, 카프카, register, connector, ksql 순으로 키는데,

register 키는 과정에서 WARNING(위에 error) 발생 합니다. ksql 쿼리를 질이 하지 않았는데 발생 합니다. 무시 해도 되는 건지 잘 모르겠습니다.

Source connect t설정이 잘 못 된 걸까여?

음, warning은 걱정하지 않으셔도 될 것 같습니다.

mysqlavro022.oc.customers 토픽에 AVRO 타입으로 값이 들어가 있는지 아래와 같이 확인해 보시겠습니까?

kafka-avro-console-consumer --bootstrap-server localhost:9092 --topic mysqlavro022.oc.customers --property schema.registry.url=http://localhost:8081 --from-beginning

네 개발자님 말씀 주신명령어를 해보았습니다.

{"customer_id":1,"email_address":"test","full_name":"test"}
{"customer_id":2,"email_address":"test2","full_name":"test2"}
{"customer_id":3,"email_address":"test3","full_name":"test3"}
{"customer_id":4,"email_address":"test4","full_name":"test4"}

출력 모습은 JSON으로 보여 지는데, avro 타입인지 search을 진행 중입니다.

 의문점이 있습니다.

말씀해 주신 명령어 쿼리시 registry_start.sh 실행한 창에 밑에와 같은 log가 출력됩니다.

[2024-02-08 19:19:10,753] INFO 127.0.0.1 - - [08/Feb./2024:10:19:10 +0000] "GET /schemas/ids/28?fetchMaxId=false&subject=null-value HTTP/1.1" 200 301 "-" "Java/11.0.21" 38 (io.confluent.rest-utils.requests:62)

위 요청으로

http GET http://localhost:8081/schemas/ids/28?fetchMaxId=false&subject=null-value 명령어를 실행 하면

아래와 같은 결과 가 나옵니다.

{
    "schema": "{
		\"type\":\"record\",
		\"name\":\"Value\",
		\"namespace\":\"mysqlavro023.oc.customers\",
		\"fields\":
			[
				{\"name\":\"customer_id\",\"type\":\"int\"},
				{\"name\":\"email_address\",\"type\":\"string\"},
				{\"name\":\"full_name\",\"type\":\"string\"}
			],
		\"connect.name\":\"mysqlavro023.oc.customers.Value\"
	      }"
}

의문점은 Value 스키마 조회는 log로 보여 지는 데, Key 스키마 조회는 log로 출력 되지 않습니다.

http GET http://localhost:8081/schemas 명령어로 조회 해 보면 key 스키마도 존재 합니다.

혹시 Key 스키마는 원래 조회를 하지 않는 건가요? 아니면 설정이 따로 필요 한건가요?

혹시 ksql-server.properties 에 ksql.schema.registry.url 값을 설정하셨나요?

ksqldb에서 schema registry에 접속하지 못하는 것 같습니다만..

ksql-server.properties 설정 정보 입니다.

listeners=http://0.0.0.0:8088
ksql.logging.processing.topic.auto.create=true
ksql.logging.processing.stream.auto.create=true
bootstrap.servers=localhost:9092
compression.type=snappy
ksql.connect.url=http://localhost:8083
ksql.schema.registry.url=http://localhost:8081
ksql.streams.state.dir=/home/user01/kafka/data/kafka-streams

혹시 잘못된 부분이나, 노친 부분이 있을까여?

음, config 설정은 문제가 없어 보입니다만,

ksqldb 강의의 실습 코드들은 아무 문제 없이 수행되었나요?

오류 내용만 보면

Cannot deserialize type struct as type int32 for path

로 되어 있는데, schema registry에는 int32로 등록되어 있는데, 실제 topic의 값은 struct여서 오류가 발생하는 것 같습니다만, kafka-avro-console-consumer 로 출력한 것 보면 customer_id가 int값으로 되어 있어서 오류 내용이 앞뒤가 맞지 않는것 같습니다.

oc_customer를 새롭게 oc_customer_stream으로 해서 table이 아니라 stream으로 다시 한번 테스트 해보시겠습니까? 토픽 이름도 새롭게 생성하고, 다시 구성해서 테스트 해보면 어떨까 싶습니다.

안녕하세요 개발자님

새해복 많이 받으세요 ㅎㅎ,

친절한 답변 감사합니다. 꼼꼼하게 챙겨 주셔거 다시한번 감동 받았습니다. 저두 멋진 개발자로 성장 해야겠다 다시한번 마음을 잡게 되었습니다. 감사합니다.

말씀주신 ksql 강의에서 제공 받은 dgen_shoe은 정상 작동 되었습니다.

또 새로운 topic을 만들어 stream 생성 작업은 여러 차례 시도 해 보았지만, 같은 오류가 발생 하였습니다.

강의 자료와 제가 하는 실습의 어떤차이가 있는지 비교 해 보았습니다.

kafka-avro-console-consumer 의 결과 구조의 차이는 없는 것으로 보여 집니다.

// dgen_shoe
{"order_id":1905,"product_id":"9952c67e-982a-4fb3-a06f-ebf10a8c36c8","customer_id":"a8602711-84d1-4306-812a-f86c7e0ae70b","ts":1609549700000}
// customers
{"customer_id":1,"email_address":"test","full_name":"test"}

스키마 구조를 보았습니다.

// dgen_shoe
    {
        "id": 37,
        "schema": "\"int\"",
        "subject": "shoe_orders_avro-key",
        "version": 1
    },
    {
        "id": 38,
        "schema": "                         {\"type\":\"record\",\"name\":\"shoe_orders\",\"namespace\":\"shoes\",
         \"fields\":[
{\"name\":\"order_id\",\"type\":\"int\"},                                   {\"name\":\"product_id\",\"type\":\"string\"},{\"name\":\"customer_id\",\"type\":\"string\"},
{\"name\":\"ts\",\"type\":{\"type\":\"long\",\"connect.version\":1,\"connect.name\":\"org.apache.kafka.connect.data.Timestamp\",\"logicalType\":\"timestamp-      millis\"}}
],
\"connect.name\":\"shoes.shoe_orders\"}",
        "subject": "shoe_orders_avro-value",
        "version": 1
    },
// customers
    {
        "id": 43,
        "schema": "        
           {\"type\":\"record\",
            \"name\":\"Key\",
            \"namespace\":\"mysqlavro028.oc.customers\",
            \"fields\":[
                    {\"name\":\"customer_id\",\"type\":\"int\"}
              ],
              \"connect.name\":\"mysqlavro028.oc.customers.Key\"}",
               "subject": "mysqlavro028.oc.customers-key",
               "version": 1
    },
    {
        "id": 44,
        "schema": "{\"type\":\"record\",
                    \"name\":\"Value\",
                    \"namespace\":\"mysqlavro028.oc.customers\",
                    \"fields\":[
{\"name\":\"customer_id\",\"type\":\"int\"},           {\"name\":\"email_address\",\"type\":\"string\"},     {\"name\":\"full_name\",\"type\":\"string\"}
],\"connect.name\":\"mysqlavro028.oc.customers.Value\"}",
        "subject": "mysqlavro028.oc.customers-value",
        "version": 1
    },

스키마 구조는 차이가 있습니다.

key 구조의 차이가 있는데, 이부분이 변경 해야된느 부분인지는 의문이 듭니다.

key에도 자세한 스키마 구조가 포함 되어야 한다고 생각은 하고 있는데, ksql에서 오류가 나는 원인이 아닐까 생각이 듭니다. 키 스키마를 변경 해서 사용 하는게 맞을까요? 맞으면 수정은 어떻게 진행 하는 건가요? 또 수정 한다면 sink connector에서는 key 스키마 참조 하여 사용 해도 문제가 없는 건가요?

제가 지금 고향에 내려와서, 지금은 좀 그렇고, 다시 집에 올라가서 직접 재현을 해보고 말씀 드려야 할 것 같습니다. 1~2 일 정도 더 걸릴것 같습니다.

넵

편하 실때 답변 주세요. 감사합니다 저두 열심 히 칮아 보겟습니다

말씀하신대로 key 타입이 문제군요.

Debezium Source Connector 는 단일 PK라도 여러 컬럼을 가정해서 기본적으로 Primitive type이 아닌 Struct Type으로 key type이 정해집니다. 예를 들어 customer_id가 pk라고 하더라도 key type은 {customer_id} 와 같은 struct type으로 스키마가 정해 줍니다.

그런데 ksqldb는 customer_id int key(또는 primary key)와 같이 정해 주면 실제 value는 struct type으로 값이 오는데, 이를 ksqldb에서 deserialize 할때 primitive int 형으로 변환시 오류가 발생합니다.

제 생각엔 debezium source connector는 pk를 struct type으로 설정하되, ksqldb에서 아래와 같이 struct type으로 인지할 수 있도록 해야 할 것 같습니다.

CREATE STREAM oc_customer_stream (

customer_id struct<customer_id int> PRIMARY KEY,

email_address varchar,

full_name varchar

) WITH (

KAFKA_TOPIC = 'mysqlavro022.oc.customers',

KEY_FORMAT = 'AVRO',

VALUE_FORMAT = 'AVRO'

);

이렇게 생성하고 select * from oc_customer_stream 을 하게 되면 customer_id가 struct type으로 값이 나오게 됩니다. customer_id를 primitive type 으로 나오게 하려면 아래와 같이 하면 됩니다.

select customer_id->customer_id, email_address, fullname from oc_customer_stream;

안녕하세요 개발자님 답변 주신 방법으로 문제를 해결 해 보았습니다.

기존 의 테이블 정보에 price 가격만 추가한 새로운 테이블을 만들고 group by를 해보고자 하였습니다.

일러주신 방법으로 Stream을 만들고 참조하여 CTAS를 만들어 보았습니다.

CTAS 조회시 아래와 같은 error가 발생 하였습니다.

Key format does not support schema.
format: KAFKA
schema: Persistence{columns=[`CUSTOMER_ID` STRUCT<`CUSTOMER_ID` BIGINT> KEY], features=[]}
reason: The 'KAFKA' format does not support type 'STRUCT'
Caused by: The 'KAFKA' format does not support type 'STRUCT'

Key 타입을 Json으로 변경후 Sink Connector에서 STM를 적용 해 보았지만 Erro만 발생 하였습니다.

+----------------------------------+-----+
|CUSTOMER_ID                       |AVG_PRICE|
+----------------------------------+-----+
|{CUSTOMER_ID=4}                   |400.0    |

extract_key 적용 해 보았습니다.

converter를 적용도 해보고, 제외도 해보았습니다. 해결 하지 못하엿습니다.

"transforms": "extract_key",      
        "transforms.extract_key.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
        "transforms.extract_key.field": "customer_id"

CREATE STREAM oc_customer_price_stream (
  customer_id struct<customer_id bigint> KEY,
  email_address varchar,
  full_name varchar,
  price int
) WITH (
  KAFKA_TOPIC = 'customers_price.oc.customers_price',	
  KEY_FORMAT = 'AVRO',
  VALUE_FORMAT = 'AVRO'
);

// customers_price.oc.customers_price의 topic도 partition을 3 개로 마췄습니다.

create table oc_customer_price_table_mv03
with (
	KAFKA_TOPIC = 'customers_price_mv_03',
	KEY_FORMAT = 'JSON', 
	VALUE_FORMAT = 'JSON',
	PARTITIONS = 3
)
as
select customer_id, avg(price) as avg_price
from oc_customer_price_stream group by customer_id;


{
    "name": "mysql_jdbc_om_sink_customers_price_sink_07",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
        "tasks.max": "1",
        "topics": "customers_price_mv_03",
        "connection.url": "jdbc:mysql://192.168.0.38/oc_sink",
        "connection.user": "cnt_dev_sink",
        "connection.password": "1234",
        "table.name.format": "oc_sink.sink_customers_price",
        "insert.mode": "upsert",
        "pk.fields": "customer_id",
        "pk.mode": "record_key",
        "delete.enabled": "true",

        "auto.evolve": "false",        
        
		
	"transforms": "extract_key",        
        "transforms.extract_key.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
        "transforms.extract_key.field": "customer_id"
    }
}

ksqldb 사용시 Debezium Source Connector에서 특별히 설정을 해야 하는 것이 있나요?

아니면 위에 같은 error를 어떻게 해결 해야 하나요?

수동으로 키 스키마를 강의에서 제공해 주는 dgen_shoe 형태로 변경 하는 방법을 사용 해도 되는 건가요?

MVIEW Group by CTAS 에서 Group by 컬럼이 Struct type이면 KEY_FORMAT이 'KAFKA' 는 안됩니다. 'KAFKA'는 Group by 컬럼이 primitive type이 되어야 합니다.

두가지 방법이 있는데, KEY_FORMAT을 KAFKA로 하려면 아래와 같이 customer_id->customer_id 와 같이 변경을 해주면 됩니다.

create table customer_stream_mview_primitive

with (

KAFKA_TOPIC = 'customer_stream_mview01',

KEY_FORMAT = 'KAFKA',

VALUE_FORMAT = 'JSON',

PARTITIONS = 1

)

as

select customer_id->customer_id as customer_id, COUNT(*) as cnt

from oc_customer_stream group by customer_id->customer_id;

KEY_FORMAT을 JSON으로 하면 아래와 같이 수행할 수도 있습니다.

create table customer_stream_mview_struct

with (

KAFKA_TOPIC = 'customer_stream_mview02',

KEY_FORMAT = 'JSON',

VALUE_FORMAT = 'JSON',

PARTITIONS = 1

)

as

select customer_id, COUNT(*) as avg_point

from oc_customer_stream group by customer_id;

안녕하세요 개발자님 , 답변 주셔서 해결 할 수 있었습니다. 감사합니다.