下面是我从Kafka主题的示例日志行
2019-03-04T08:53:03.023Z "cd8cbe" 100.212.212.212 - - [20/Feb/2019:12:13:33 +0000] "GET http://dl-mysite.com/drm/PRIORITY1080/HINDI_MOVIES/somemovie.mp4/video/avc1/4/seg-1281.m4s HTTP/1.1" 200 325040 "-" "Dalvik/2.1.0 (Linux; U; Android 6.0; Le X509 Build/DHXOSOP5801911241S)" "256" "0.000"
我正在尝试使用ksql在上面的主题上创建流。下面是我用来创建流的脚本。在运行下面的流创建脚本之后,它返回“stream created”消息,但是select语句(select*fromtest\u duplicate\u stream;)什么都不退。
CREATE STREAM test_duplicate_stream (logArrivalTime varchar,edgeid varchar,ip varchar,col1_empty varchar,col2_empty varchar, eventdate varchar,url varchar,response_code int,response_length BIGINT,col3_empty varchar,user_agent varchar,request_length varchar, response_time varchar) WITH (kafka_topic='test_duplicate',VALUE_FORMAT='DELIMITED');
我认为“delimited”不是正确的值,因为我的字段不是逗号分隔的,而是空格分隔的。为我的日志行创建流的正确方法是什么?
1条答案
按热度按时间ctehm74n1#
ksql目前支持:
json文件
逗号分隔(分隔)
阿夫罗
如果您的数据不是上述格式中的一种,那么您将无法在不首先更改序列化的情况下对其进行处理。
另请参阅ksql疑难解答的注解,以了解查询何时不返回数据。