storm-sql-kafka问题情况

时间:2023-03-09 03:06:27
storm-sql-kafka问题情况

  首先上官方文档:http://storm.apache.org/releases/1.2.2/storm-sql.html

解决的问题

  1、kafka版本不对

  开始测试时采用storm1.2.2+kafka2.0.1的组合进行测试,测试过程完全按照官方文档所写进行。出现了kafka无法获取scala依赖的问题,经查找发现是storm-kafka采用的是2.10版本的scala,而kafka2.0.1采用的是2.11或者2.12,无法兼容。于是只能换回kafka0.8.2.2进行测试。

  2、storm找不到依赖

  按照官方文档所写的提交命令无法找到storm-sql-runtime,所以需要加入storm-sql相关依赖包,任务提交命令如下:

  

./bin/storm sql /softwares/storm-1.2./storm-sql.sql stormsql1 --artifacts "org.apache.storm:storm-sql-kafka:1.2.2,org.apache.storm:storm-kafka:1.2.2,org.apache.kafka:kafka_2.10:0.8.2.2^org.slf4j:slf4j-log4j12,org.apache.kafka:kafka-clients:0.8.2.2,org.apache.storm:storm-sql-runtime:1.2.2,org.apache.storm:storm-sql-core:1.2.2"

遇到的问题

  任务提交成功,但是bolt执行总是失败。

  此问题无法直接解决,并且在UI上无法找到日志。

  排除过程如下:首先去日志里看看:

less logs/workers-artifacts/stormsql1--//worker.log

  发现:

-- ::03.992 o.a.s.t.t.s.TransactionalState Thread--$spoutcoord-spout-TRIDENTSTREAMSCANREL_21_0-executor[ ] [WARN] Failed to deserialize zookeeper data for path /meta/
org.apache.storm.shade.org.json.simple.parser.ParseException: null
at org.apache.storm.shade.org.json.simple.parser.Yylex.yylex(Unknown Source) ~[storm-core-1.2..jar:1.2.]
at org.apache.storm.shade.org.json.simple.parser.JSONParser.nextToken(Unknown Source) ~[storm-core-1.2..jar:1.2.]
at org.apache.storm.shade.org.json.simple.parser.JSONParser.parse(Unknown Source) ~[storm-core-1.2..jar:1.2.]
at org.apache.storm.shade.org.json.simple.parser.JSONParser.parse(Unknown Source) ~[storm-core-1.2..jar:1.2.]
at org.apache.storm.shade.org.json.simple.parser.JSONParser.parse(Unknown Source) ~[storm-core-1.2..jar:1.2.]
at org.apache.storm.shade.org.json.simple.JSONValue.parseWithException(Unknown Source) ~[storm-core-1.2..jar:1.2.]
at org.apache.storm.trident.topology.state.TransactionalState.getData(TransactionalState.java:) [storm-core-1.2..jar:1.2.]
at org.apache.storm.trident.topology.state.RotatingTransactionalState.sync(RotatingTransactionalState.java:) [storm-core-1.2..jar:1.2.]
at org.apache.storm.trident.topology.state.RotatingTransactionalState.<init>(RotatingTransactionalState.java:) [storm-core-1.2..jar:1.2.]
at org.apache.storm.trident.spout.TridentSpoutCoordinator.prepare(TridentSpoutCoordinator.java:) [storm-core-1.2..jar:1.2.]
at org.apache.storm.topology.BasicBoltExecutor.prepare(BasicBoltExecutor.java:) [storm-core-1.2..jar:1.2.]
at org.apache.storm.daemon.executor$fn__10795$fn__10808.invoke(executor.clj:) [storm-core-1.2..jar:1.2.]
at org.apache.storm.util$async_loop$fn__553.invoke(util.clj:) [storm-core-1.2..jar:1.2.]
at clojure.lang.AFn.run(AFn.java:) [clojure-1.7..jar:?]
at java.lang.Thread.run(Thread.java:) [?:1.8.0_131]
-- ::03.997 o.a.s.d.executor Thread--$spoutcoord-spout-TRIDENTSTREAMSCANREL_21_0-executor[ ] [INFO] Prepared bolt $spoutcoord-spout-TRIDENTSTREAMSCANREL_21_0:()
-- ::04.550 o.a.s.k.t.TridentKafkaState Thread--b--executor[ ] [WARN] Could not send messages [[null, null], [null, null], [null, null]] to topic = testout
java.lang.NullPointerException: null
at org.apache.storm.kafka.IntSerializer.serialize(IntSerializer.java:) ~[dep-org.apache.storm-storm-kafka-jar-1.2..jar.:1.2.]
at org.apache.storm.kafka.IntSerializer.serialize(IntSerializer.java:) ~[dep-org.apache.storm-storm-kafka-jar-1.2..jar.:1.2.]
at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:) ~[dep-org.apache.kafka-kafka-clients-jar-0.8.2.2.jar.:?]
at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:) ~[dep-org.apache.kafka-kafka-clients-jar-0.8.2.2.jar.:?]
at org.apache.storm.kafka.trident.TridentKafkaState.updateState(TridentKafkaState.java:) [dep-org.apache.storm-storm-kafka-jar-1.2..jar.:1.2.]
at org.apache.storm.kafka.trident.TridentKafkaUpdater.updateState(TridentKafkaUpdater.java:) [dep-org.apache.storm-storm-kafka-jar-1.2..jar.:1.2.]
at org.apache.storm.kafka.trident.TridentKafkaUpdater.updateState(TridentKafkaUpdater.java:) [dep-org.apache.storm-storm-kafka-jar-1.2..jar.:1.2.]
at org.apache.storm.trident.planner.processor.PartitionPersistProcessor.finishBatch(PartitionPersistProcessor.java:) [storm-core-1.2..jar:1.2.]
at org.apache.storm.trident.planner.SubtopologyBolt.finishBatch(SubtopologyBolt.java:) [storm-core-1.2..jar:1.2.]
at org.apache.storm.trident.topology.TridentBoltExecutor.finishBatch(TridentBoltExecutor.java:) [storm-core-1.2..jar:1.2.]
at org.apache.storm.trident.topology.TridentBoltExecutor.checkFinish(TridentBoltExecutor.java:) [storm-core-1.2..jar:1.2.]
at org.apache.storm.trident.topology.TridentBoltExecutor.execute(TridentBoltExecutor.java:) [storm-core-1.2..jar:1.2.]
at org.apache.storm.daemon.executor$fn__10795$tuple_action_fn__10797.invoke(executor.clj:) [storm-core-1.2..jar:1.2.]
at org.apache.storm.daemon.executor$mk_task_receiver$fn__10716.invoke(executor.clj:) [storm-core-1.2..jar:1.2.]
at org.apache.storm.disruptor$clojure_handler$reify__10135.onEvent(disruptor.clj:) [storm-core-1.2..jar:1.2.]
at org.apache.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:) [storm-core-1.2..jar:1.2.]
at org.apache.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:) [storm-core-1.2..jar:1.2.]
at org.apache.storm.disruptor$consume_batch_when_available.invoke(disruptor.clj:) [storm-core-1.2..jar:1.2.]
at org.apache.storm.daemon.executor$fn__10795$fn__10808$fn__10861.invoke(executor.clj:) [storm-core-1.2..jar:1.2.]
at org.apache.storm.util$async_loop$fn__553.invoke(util.clj:) [storm-core-1.2..jar:1.2.]
at clojure.lang.AFn.run(AFn.java:) [clojure-1.7..jar:?]
at java.lang.Thread.run(Thread.java:) [?:1.8.0_131]

  接下来分为两步,其一是查找/meta/1这个目录在zookeeper的哪里,另一个是查找报错的源码在哪里,根据storm源码可以看到:

  

    public Object getData(String path) {
path = "/" + path;
try {
Object data;
if (_curator.checkExists().forPath(path) != null) {
// Use parseWithException instead of parse so we can capture deserialization errors in the log.
// They are likely to be bugs in the spout code.
try {
data = JSONValue.parseWithException(new String(_curator.getData().forPath(path), "UTF-8"));
} catch (ParseException e) {
LOG.warn("Failed to deserialize zookeeper data for path {}", path, e);
data = null;
}
} else {
data = null;
}
LOG.debug("Get. [path = {}] => [data = {}]", path, data);
return data;
} catch (Exception e) {
throw new RuntimeException(e);
}
}

  可以看到报错的是JSONValue.parseWithException出错。获取数据是解析一个json,并放到data里,而通过判断条件我们可以知道path是有内容的,只不过读不出来。

  通过查询源码找到/meta/1的完整路径为:/transactional/....../meta/1,去此路径下找到对应的文件为:/transactional/TRIDENTSTREAMSCANREL_21_0/coordinator/meta/1

  获取此文件信息:get /transactional/TRIDENTSTREAMSCANREL_21_0/coordinator/meta/1

  

[GlobalPartitionInformation{topic=testin, partitionMap={=cloud22:, =cloud22:, =cloud22:, =cloud22:, =cloud22:}}]
cZxid = 0xf02f
ctime = Thu Nov :: CST
mZxid = 0x45a9b
mtime = Thu Nov :: CST
pZxid = 0xf02f
cversion =
dataVersion =
aclVersion =
ephemeralOwner = 0x0
dataLength =
numChildren =

  可以发现,emmm...这一段信息看格式应该是通过toString写出来的,但是读却是用json格式读。代码鬼才,我觉得不行。

结论:storm-sql-kafka基本不可用,至少未找到解决方案,从错误来看,zookeeper中写kafka partitions meta的代码和读的代码完全无法匹配上。在不修改源码的前提下无法解决此问题。