为 warc 螺栓设置新流失败

Question

我正在尝试设置新流以将 Tika 螺栓连接到 warc 螺栓。

import com.digitalpebble.stormcrawler.tika.ParserBolt;
import com.digitalpebble.stormcrawler.warc.WARCHdfsBolt;

builder.setBolt("tika", new ParserBolt(), numWorkers)
  .localOrShuffleGrouping("shunt","tika");

WARCHdfsBolt warcbolt = getWarcBolt("XX");

builder.setBolt("warc", warcbolt, numWorkers)
  .localOrShuffleGrouping("tika",  "warc");

在 Tika 定义中，我修改了 outputDeclarerFields 函数，以定义我的新 "warc" 流：

@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
  declarer.declare(new Fields("url", "content", "metadata", "text"));
  declarer.declareStream(StatusStreamName, new Fields("url", "metadata", "status"));
  declarer.declareStream("warc",   new Fields("url", "content", "metadata", "text"));
}

但是，当我在本地模式下启动拓扑时，我得到：

14308 [main] WARN o.a.s.d.s.Slot - SLOT debian8:1027 Starting in state EMPTY - assignment null 14308 [main] WARN o.a.s.d.s.Slot - SLOT debian8:1028 Starting in state EMPTY - assignment null 14308 [main] WARN o.a.s.d.s.Slot - SLOT debian8:1029 Starting in state EMPTY - assignment null 14309 [main] INFO o.a.s.l.AsyncLocalizer - Cleaning up unused topologies in /tmp/a1e3b7f5-e251-40ae-a032-b0839ca103c8/supervisor/stormdist 14318 [main] INFO o.a.s.d.s.Supervisor - Starting supervisor with id f42c64cd-7c36-40ab-9f85-4b7751ed2d6a at host debian8. 15030 [main] WARN o.a.s.d.nimbus - Topology submission exception. (topology name='xxCrawler') #error { :cause nil :via [{:type org.apache.storm.generated.InvalidTopologyException :message nil
:at [org.apache.storm.daemon.common$validate_structure_BANG_ invoke common.clj 185]}] :trace [[org.apache.storm.daemon.common$validate_structure_BANG_ invoke common.clj 185]
[org.apache.storm.daemon.common$system_topology_BANG_ invoke common.clj 378]
[org.apache.storm.daemon.nimbus$mk_reified_nimbus$reify__10782 submitTopologyWithOpts nimbus.clj 1694]
[org.apache.storm.daemon.nimbus$mk_reified_nimbus$reify__10782 submitTopology nimbus.clj 1726]
[sun.reflect.NativeMethodAccessorImpl invoke0 NativeMethodAccessorImpl.java -2]
[sun.reflect.NativeMethodAccessorImpl invoke NativeMethodAccessorImpl.java 62]
[sun.reflect.DelegatingMethodAccessorImpl invoke DelegatingMethodAccessorImpl.java 43] [java.lang.reflect.Method invoke Method.java 498] [clojure.lang.Reflector invokeMatchingMethod Reflector.java 93] [clojure.lang.Reflector invokeInstanceMethod Reflector.java 28] [org.apache.storm.testing$submit_local_topology invoke testing.clj 310]
[org.apache.storm.LocalCluster$_submitTopology invoke LocalCluster.clj 49] [org.apache.storm.LocalCluster submitTopology nil -1]
[com.digitalpebble.stormcrawler.ConfigurableTopology submit ConfigurableTopology.java 76]
[com.digitalpebble.stormcrawler.ConfigurableTopology submit ConfigurableTopology.java 65] [xx.xx.xx.xx.xxTopology run xxTopology.java 111]
[com.digitalpebble.stormcrawler.ConfigurableTopology start ConfigurableTopology.java 50] [xx.xx.xx.xx.xxTopology main xxTopology.java 53]]} 15035 [main] ERROR o.a.s.s.o.a.z.s.NIOServerCnxnFactory - Thread Thread[main,5,main] died org.apache.storm.generated.InvalidTopologyException: null at org.apache.storm.daemon.common$validate_structure_BANG_.invoke(common.clj:185) ~[storm-core-1.1.0.jar:1.1.0] at org.apache.storm.daemon.common$system_topology_BANG_.invoke(common.clj:378) ~[storm-core-1.1.0.jar:1.1.0] at org.apache.storm.daemon.nimbus$mk_reified_nimbus$reify__10782.submitTopologyWithOpts(nimbus.clj:1694) ~[storm-core-1.1.0.jar:1.1.0] at org.apache.storm.daemon.nimbus$mk_reified_nimbus$reify__10782.submitTopology(nimbus.clj:1726) ~[storm-core-1.1.0.jar:1.1.0] at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_131] at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:1.8.0_131] at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_131] at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_131] at clojure.lang.Reflector.invokeMatchingMethod(Reflector.java:93) ~[clojure-1.7.0.jar:?] at clojure.lang.Reflector.invokeInstanceMethod(Reflector.java:28) ~[clojure-1.7.0.jar:?] at org.apache.storm.testing$submit_local_topology.invoke(testing.clj:310) ~[storm-core-1.1.0.jar:1.1.0] at org.apache.storm.LocalCluster$_submitTopology.invoke(LocalCluster.clj:49) ~[storm-core-1.1.0.jar:1.1.0] at org.apache.storm.LocalCluster.submitTopology(Unknown Source) ~[storm-core-1.1.0.jar:1.1.0] at com.digitalpebble.stormcrawler.ConfigurableTopology.submit(ConfigurableTopology.java:76) ~[xx-crawler-1.1.jar:?] at com.digitalpebble.stormcrawler.ConfigurableTopology.submit(ConfigurableTopology.java:65) ~[xx-1.1.jar:?] at xx.xx.xx.xx.xxTopology.run(xxTopology.java:111) ~[xx-crawler-1.1.jar:?] at com.digitalpebble.stormcrawler.ConfigurableTopology.start(ConfigurableTopology.java:50) ~[xx-crawler-1.1.jar:?] at xx.xx.xx.xx.xxTopology.main(xxTopology.java:53) ~[xx-crawler-1.1.jar:?]

非常感谢任何帮助！！

请注意，如果我使用 StatusStreamName ("status") 流连接 tika 和 warc 螺栓，它工作正常。

谢谢，

艾蒂安

Answer 1

WARC 是从未经解析的原始内容生成的。您应该将 WARC 连接到 Fetcher 的输出而不是 Parser bolt。

您不需要仅为 warc 声明新流，您可以简单地将 warc 螺栓连接到来自 Tika 螺栓的默认流。

我在你的代码中看到了

import com.digitalpebble.stormcrawler.tika.ParserBolt;

这表明您依赖于默认实现（它不会生成 'warc' 流）。您是否忘记用修改后的实现替换它？

为 warc 螺栓设置新流失败

Setting up a new stream for warc bolt fails

web-crawler

stormcrawler