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

죄송합니다 급하게 글을 쓰느라 정확하지 못하였습니다.
다시 정리하자면 source connector로 받아서 넣은 원본 stream은 raw_orders입니다.

select * from raw_orders;
+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+
|ORDER_ID            |ORDER_DATETIME      |CUSTOMER_ID         |ORDER_STATUS        |PRICE               |STORE_ID            |
+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+
|{ORDER_ID=1}        |2024-06-17T00:00:00Z|1001                |Completed           |59.99               |1                   |
|{ORDER_ID=2}        |2024-06-17T01:15:00Z|1002                |Pending             |29.99               |2                   |
|{ORDER_ID=3}        |2024-06-17T02:30:00Z|1003                |Cancelled           |19.99               |3                   |
|{ORDER_ID=4}        |2024-06-17T06:30:00Z|1004                |Cancelled           |190.99              |3                   |
|{ORDER_ID=5}        |2024-06-16T06:30:00Z|1004                |Cancelled           |190.99              |3                   |
|{ORDER_ID=6}        |2024-06-16T06:30:00Z|1006                |yes                 |1190.99             |6                   |
|{ORDER_ID=7}        |2024-06-18T00:20:30Z|1007                |yes                 |1190.99             |7                   |

총 7개의 데이터를 넣었습니다.

이 stream을 기반으로 하여 10분이 지나면 table로 가져오게끔 만들었습니다

CREATE TABLE timeout WITH (KAFKA_TOPIC='timeout' , KEY_FORMAT='AVRO', VALUE_FORMAT='AVRO', PARTITIONS=1) AS
> SELECT
> order_id -> order_id AS  order_id,
> TIMESTAMPADD(MILLISECONDS, 9 * 3600 * 1000, PARSE_TIMESTAMP(LATEST_BY_OFFSET(order_datetime), 'yyyy-MM-dd''T''HH:mm:ssX')) AS last_log_time
> FROM raw_orders
> GROUP BY order_id -> order_id
> HAVING ((UNIX_TIMESTAMP(CONVERT_TZ(FROM_UNIXTIME(UNIX_TIMESTAMP()), 'UTC', 'Asia/Seoul')) - UNIX_TIMESTAMP(TIMESTAMPADD(MILLISECONDS, 9 * 3600 * 1000, PARSE_TIMESTAMP(LATEST_BY_OFFSET(order_datetime), 'yyyy-MM-dd''T''HH:mm:ssX')))) / 1000 > 600)
> EMIT CHANGES;

1-6번까지의 데이터는 이미 설정해둔 시간이 지나서 table에는 담겨있습니다.

 select * from timeout;
+----------------------------------------------------------------+----------------------------------------------------------------+
|ORDER_ID                                                        |LAST_LOG_TIME                                                   |
+----------------------------------------------------------------+----------------------------------------------------------------+
|1                                                               |2024-06-17T09:00:00.000                                         |
|2                                                               |2024-06-17T10:15:00.000                                         |
|3                                                               |2024-06-17T11:30:00.000                                         |
|4                                                               |2024-06-17T15:30:00.000                                         |
|5                                                               |2024-06-16T15:30:00.000                                         |
|6                                                               |2024-06-16T15:30:00.000                                         |

하지만 7번 데이터가 들어오고 10분이 지나서 table에는 7번 데이터가 담기는데
topic에는 담기지 않습니다.

select * from timeout;
+----------------------------------------------------------------+----------------------------------------------------------------+
|ORDER_ID                                                        |LAST_LOG_TIME                                                   |
+----------------------------------------------------------------+----------------------------------------------------------------+
|1                                                               |2024-06-17T09:00:00.000                                         |
|2                                                               |2024-06-17T10:15:00.000                                         |
|3                                                               |2024-06-17T11:30:00.000                                         |
|4                                                               |2024-06-17T15:30:00.000                                         |
|5                                                               |2024-06-16T15:30:00.000                                         |
|6                                                               |2024-06-16T15:30:00.000                                         |
|7                                                               |2024-06-18T09:20:30.000                                         |
Query terminated
ksql> print timeout;
Key format: AVRO or KAFKA_STRING
Value format: AVRO
rowtime: 2024/06/17 01:51:02.907 Z, key: 1, value: {"LAST_LOG_TIME": 1718614800000}, partition: 0
rowtime: 2024/06/17 01:51:02.922 Z, key: 2, value: {"LAST_LOG_TIME": 1718619300000}, partition: 0
rowtime: 2024/06/17 02:19:38.295 Z, key: 5, value: {"LAST_LOG_TIME": 1718551800000}, partition: 0
rowtime: 2024/06/18 00:24:06.884 Z, key: 6, value: {"LAST_LOG_TIME": 1718551800000}, partition: 0
^CTopic printing ceased

참고로 3,4번도 같은 방식으로 진행했지만 담기지 않았습니다.

어떻게하면 좋을까요?

원본 stream 명은 raw_orders인데, 아래 CTAS를 보면 SELECT를 orders에서 하는데, orders는 어떤 거인지요?

Create table timeout

as

select ...

from orders

죄송합니다 오타가 있었습니다 raw_orders에서 받아온게 맞습니다

질문글에 오타입니다. 진행은 raw_orders로 했습니다. 시간이 지나고 table에는 담기지만
topic에는 담기지않습니다..

음, 이건 원인을 정확히 모르겠군요.

먼저 아래와 같이

0. 토픽명을 바꿔 보셨나요? 헷갈릴수 있으니 토픽명을 일단 변경해 보시지요. 그리고

1. CTAS에서 Having절을 삭제해 보시고, 다시 한번 테스트 해보시지요.

2. 1번이 마찬가지면 일단 consumer로 데이터를 확인해 보시지요.

    

   kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic 토픽명 --from-beginning