CouchBasePojoseOperator在运行ApacheApex应用程序时导致启动问题

5t7ly7z5  于 2021-06-07  发布在  Kafka
关注(0)|答案(0)|浏览(288)

我有一个基本的顶点流，Kafka输入操作符输入couchbase集操作符。Kafka方面的工作很好，并证明了它删除沙发经营者。我试过切换malhar库的版本，看看它是否在最新版本中被破坏。
我还使用了apex的datatorrent风格。
当我添加couch操作符时，我得到以下异常

  1. java.lang.RuntimeException: Error creating local cluster
  3. at org.apache.apex.engine.EmbeddedAppLauncherImpl.getController(EmbeddedAppLauncherImpl.java:124)
  4. at org.apache.apex.engine.EmbeddedAppLauncherImpl.getController(EmbeddedAppLauncherImpl.java:46)
  5. at com.usaa.data.streams.apex.ApplicationTest.testApplication(ApplicationTest.java:30)
  6. at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  7. at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
  8. at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  9. at java.lang.reflect.Method.invoke(Method.java:498)
  10. at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:45)
  11. at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:15)
  12. at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:42)
  13. at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:20)
  14. at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:263)
  15. at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:68)
  16. at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:47)
  17. at org.junit.runners.ParentRunner$3.run(ParentRunner.java:231)
  18. at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:60)
  19. at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:229)
  20. at org.junit.runners.ParentRunner.access$000(ParentRunner.java:50)
  21. at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:222)
  22. at org.junit.runners.ParentRunner.run(ParentRunner.java:300)
  23. at org.junit.runner.JUnitCore.run(JUnitCore.java:157)
  24. at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
  25. at com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:51)
  26. at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:237)
  27. at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)
  28. Caused by: com.esotericsoftware.kryo.KryoException: Class cannot be created (missing no-arg constructor): org.codehaus.jackson.map.DeserializationConfig
  29. Serialization trace:
  30. _deserializationConfig (org.codehaus.jackson.map.ObjectMapper)
  31. mapper (com.datatorrent.contrib.couchbase.CouchBaseJSONSerializer)
  32. serializer (com.datatorrent.contrib.couchbase.CouchbasePOJOSetOperator)
  33. at com.esotericsoftware.kryo.Kryo$DefaultInstantiatorStrategy.newInstantiatorOf(Kryo.java:1228)
  34. at com.esotericsoftware.kryo.Kryo.newInstantiator(Kryo.java:1049)
  35. at com.esotericsoftware.kryo.Kryo.newInstance(Kryo.java:1058)
  36. at com.esotericsoftware.kryo.serializers.FieldSerializer.create(FieldSerializer.java:547)
  37. at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:523)
  38. at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)
  39. at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:106)
  40. at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
  41. at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)
  42. at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:106)
  43. at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
  44. at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)
  45. at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:106)
  46. at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
  47. at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:761)
  48. at com.datatorrent.common.util.FSStorageAgent.retrieve(FSStorageAgent.java:192)
  49. at com.datatorrent.stram.plan.logical.LogicalPlan$OperatorMeta.readObject(LogicalPlan.java:898)
  50. at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  51. at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
  52. at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  53. at java.lang.reflect.Method.invoke(Method.java:498)
  54. at java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1058)
  55. at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2122)
  56. at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2013)
  57. at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1535)
  58. at java.io.ObjectInputStream.readObject(ObjectInputStream.java:422)
  59. at java.util.ArrayList.readObject(ArrayList.java:791)
  60. at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  61. at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
  62. at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  63. at java.lang.reflect.Method.invoke(Method.java:498)
  64. at java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1058)
  65. at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2122)
  66. at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2013)
  67. at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1535)
  68. at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2231)
  69. at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2155)
  70. at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2013)
  71. at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1535)
  72. at java.io.ObjectInputStream.readObject(ObjectInputStream.java:422)
  73. at com.datatorrent.stram.plan.logical.LogicalPlan.read(LogicalPlan.java:2326)
  74. at com.datatorrent.stram.StramLocalCluster.cloneLogicalPlan(StramLocalCluster.java:323)
  75. at com.datatorrent.stram.StramLocalCluster.<init>(StramLocalCluster.java:285)
  76. at org.apache.apex.engine.EmbeddedAppLauncherImpl.getController(EmbeddedAppLauncherImpl.java:122)
  77. ... 24 more

以下是相关的应用程序代码：

  1. KafkaSinglePortInputOperator kafkaInput = dag.addOperator("kafkaInput", KafkaSinglePortInputOperator.class);
  2.     kafkaInput.setTopics("testing2");
  3.     Properties props = new Properties();
  4.     props.put("bootstrap.servers", "localhost:9092");
  5.     props.put("group.id", "test");
  6.     props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  7.     props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  8.     props.put("enable.auto.commit", "true");
  9.     kafkaInput.setConsumerProps(props);
  10.     kafkaInput.setClusters("localhost:9092");
  12.     CouchbasePOJOSetOperator couchOutput = dag.addOperator("couchOutput", CouchbasePOJOSetOperator.class);
  14.     CouchBaseWindowStore store = new CouchBaseWindowStore();
  15.     store.setBucket("default");
  16.     store.setUriString("localhost:8091,localhost:8091");
  17.     store.setUserConfig("");
  18.     store.setPasswordConfig("");
  19.     store.setPassword("");
  20.     couchOutput.setStore(store);
  21.     try{
  22.         store.connect();
  23.     }catch (Exception e){
  24.         e.printStackTrace();
  25.     }
  27.     CouchBaseJSONSerializer serializer = new CouchBaseJSONSerializer();
  28.     couchOutput.setSerializer(serializer);
  29.     ArrayList<String> expressions = new ArrayList<String>();
  30.     expressions.add("getValue()");
  31.     couchOutput.setExpressions(expressions);
  33.     dag.addStream("kafkaInput", kafkaInput.outputPort, couchOutput.input).setLocality(Locality.CONTAINER_LOCAL);
Javaapache-kafkacouchbaseapache-apex

暂无答案！

目前还没有任何答案，快来回答吧！

我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com