- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
我有一个原型(prototype) Apache Beam 管道,我尝试使用以下配置从 RabbitMQ 读取数据
p.apply("read_from_rabbit", RabbitMqIO.read()
.withUri(options.getRabbitMQUri())
.withQueue(options.getRabbitMQQueue())
)
.apply("extract_json_data", MapElements.via(new RabbitMessageToKafkaMessage()))
当我尝试运行它时,我总是得到
Exception in thread "main" java.lang.NoClassDefFoundError: com/rabbitmq/client/QueueingConsumer$Delivery
at java.lang.Class.getDeclaredConstructors0(Native Method)
at java.lang.Class.privateGetDeclaredConstructors(Class.java:2671)
at java.lang.Class.getDeclaredConstructors(Class.java:2020)
at java.io.ObjectStreamClass.computeDefaultSUID(ObjectStreamClass.java:1793)
at java.io.ObjectStreamClass.access$100(ObjectStreamClass.java:72)
at java.io.ObjectStreamClass$1.run(ObjectStreamClass.java:253)
at java.io.ObjectStreamClass$1.run(ObjectStreamClass.java:251)
at java.security.AccessController.doPrivileged(Native Method)
at java.io.ObjectStreamClass.getSerialVersionUID(ObjectStreamClass.java:250)
at java.io.ObjectStreamClass.writeNonProxy(ObjectStreamClass.java:735)
at java.io.ObjectOutputStream.writeClassDescriptor(ObjectOutputStream.java:668)
at java.io.ObjectOutputStream.writeNonProxyDesc(ObjectOutputStream.java:1282)
at java.io.ObjectOutputStream.writeClassDesc(ObjectOutputStream.java:1231)
at java.io.ObjectOutputStream.writeClass(ObjectOutputStream.java:1213)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1120)
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
at org.apache.beam.sdk.util.SerializableUtils.serializeToByteArray(SerializableUtils.java:51)
at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.CoderTranslation.toCustomCoder(CoderTranslation.java:119)
at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.CoderTranslation.toProto(CoderTranslation.java:83)
at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.SdkComponents.registerCoder(SdkComponents.java:250)
at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.PCollectionTranslation.toProto(PCollectionTranslation.java:35)
at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.SdkComponents.registerPCollection(SdkComponents.java:205)
at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.PTransformTranslation.translateAppliedPTransform(PTransformTranslation.java:369)
at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.ParDoTranslation$ParDoTranslator.translate(ParDoTranslation.java:120)
at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.PTransformTranslation.toProto(PTransformTranslation.java:149)
at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.ParDoTranslation.getParDoPayload(ParDoTranslation.java:651)
at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.ParDoTranslation.isSplittable(ParDoTranslation.java:666)
at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.PTransformMatchers$6.matches(PTransformMatchers.java:269)
at org.apache.beam.sdk.Pipeline$2.visitPrimitiveTransform(Pipeline.java:280)
at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:665)
at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:657)
at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:657)
at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:657)
at org.apache.beam.sdk.runners.TransformHierarchy$Node.access$600(TransformHierarchy.java:317)
at org.apache.beam.sdk.runners.TransformHierarchy.visit(TransformHierarchy.java:251)
at org.apache.beam.sdk.Pipeline.traverseTopologically(Pipeline.java:458)
at org.apache.beam.sdk.Pipeline.replace(Pipeline.java:258)
at org.apache.beam.sdk.Pipeline.replaceAll(Pipeline.java:208)
at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:154)
at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:64)
at org.apache.beam.sdk.Pipeline.run(Pipeline.java:313)
at org.apache.beam.sdk.Pipeline.run(Pipeline.java:299)
at myCompany.myProject.RabbitToKafka.runTransformer(RabbitToKafka.java:54)
at myCompany.myProject.RabbitToKafka.main(RabbitToKafka.java:61)
Caused by: java.lang.ClassNotFoundException: com.rabbitmq.client.QueueingConsumer$Delivery
at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:335)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
... 48 more
我的理解是,由于我的 RabbitMessageToKafkaMessage
类读取 RabbitMQ 消息,这些消息又包含 RabbitMQ 数据,特别是注入(inject)到构造函数中的传递信息: public RabbitMqMessage(StringroutingKey, QueueingConsumer.Delivery交货){
这样可以吗?
如果是这样,如何在读取操作期间将我的 RabbitMQMessage
转换为 KV
?
EDIT 1 从 Eclipse 运行管道时发生错误。
编辑 2 该项目是使用 Eclipse 运行的 Maven 项目。我的 Apache Bean 依赖项都是 2.12.0 版本(这是最新的)。
我的依赖树如下(至少是关于RabbitMQ的部分)
myCompany:myProject:jar:0.1.5-SNAPSHOT
+- org.apache.beam:beam-sdks-java-io-rabbitmq:jar:2.12.0:compile
| \- com.rabbitmq:amqp-client:jar:5.4.3:compile
最佳答案
看来,将我以前的 pom 与 Apache Beam 兼容的 pom 合并不是一个好主意,因为它在两个不同版本的 amqp-client 之间造成了冲突。删除 Spring boot 父 pom 通过删除错误的 amqp-client 版本解决了该错误。
关于java - 如何使用 Apache Beam 从 RabbitMQ 读取数据,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/55955978/
我正在使用 Apache Beam 从 Kafka 读取数据。由于乱序写入 Kafka,我想使用有效负载中的事件时间戳而不是默认的 LogAppendTime。 我看到了一个解决方案 Apache B
我有一个通用输入请求,其中包含需要转换和保存的输入。如果需要转换生成的输出,我会为它实现一个新的处理器(转换器)。 class Request { Input input; Transform
我已经在 Apache Beam 上工作了几天。我想快速迭代我正在工作的应用程序,并确保我正在构建的管道没有错误。在 Spark 中我们可以使用 sc.parallelise当我们应用一些 Actio
我有一个简单的波束管道,它使用带有 tf 变换的通用句子编码器获取一些文本并获得嵌入。与使用 tf 1 制作的演示非常相似。 import tensorflow as tf import apache
使用Apache Beam丰富数据时,对每个数据项都进行一次API调用会不会出错? (我是 Apache Beam 的新手) 最佳答案 不,但您可以批处理 API 调用以获得更好的性能。查看 this
我的任务是为一款应用添加支持,以便通过 Android 上的“NFC”在设备之间传输大型数据文件(数十兆字节)。 我知道 Android 上真正的 NFC 非常慢,但我知道 ICS 支持将批量数据传输
NFC:S-beam 和 Android beam 有什么区别?有人可以解释 Wifi-Direct/Bluetooth 激活和传输数据的确切流程吗? 最佳答案 在 stackexchanged 上解
我正在使用 Beam 管道计算流式数据的电话号码频率。我使用的滑动窗口每 5 分钟重复一次,总周期为 15 分钟,因此正如预期的那样,对于某些输入,当输入落在多个窗口中时,我会得到多个输出。 计算出现
输入的PCollection是http requests,是一个有界数据集。我想在 ParDo 中进行异步 http 调用(Java),解析响应并将结果放入输出 PCollection 中。我的代码如
输入的PCollection是http requests,是一个有界数据集。我想在 ParDo 中进行异步 http 调用(Java),解析响应并将结果放入输出 PCollection 中。我的代码如
在使用 PAssert 为我的光束管道编写单元测试时,管道输出对象很好,但在与以下断言错误进行比较时测试失败: java.lang.AssertionError: Decode pubsub mess
我正在尝试从 here 运行 Wordcount 演示与 Samza Runner。这是我的build.gradle plugins { id 'eclipse' id 'java' id
我正在尝试使用 Beam 和 Flink runner 设置流处理管道。 Flink 是一个本地 session 部署,包含以下 docker-compose 文件: version: "3" ser
在尝试编译我的 Phoenix 项目的发行版时,出现以下错误: $ mix release .... ==> Generated .appup for myapp 0.0.1 -> 0.0.2 ===
我正在尝试使用 Apache beam-dataflow 连接到安装在云实例中的配置单元实例。当我运行它时,出现以下异常。当我使用 Apache Beam 访问此数据库时,就会发生这种情况。我见过很多
我正在使用 zsh,并且我已经安装了 gcloud,以便通过我的 Mac 上的本地终端与 GCP 进行交互。我遇到了这个错误“zsh:找不到匹配项:apache-beam[gcp]”。但是,当我在 G
主要记录两种不同的beam search版本 版本一 使用类似层次遍历的方式进行搜索,用队列进行维护,每次循环对当前层的所有节点进行搜索,这些节点每个分别对应topk个节点作为下一层候选节点,取
我的目标是创建一个每秒调用后端(云托管)服务最多次数的管道......我该如何实现? 背景故事:想象一下后端服务使用单个输入调用并返回单个输出。该服务具有与其关联的配额,允许每秒最大请求数(假设每秒
我想写入一个 gs 文件,但在编译时我不知道文件名。它的名称基于在运行时定义的行为。我该如何继续? 最佳答案 如果你使用 Beam Java,你可以使用 FileIO.writeDynamic()为此
我试图弄清楚如何使用Apache Beam读取大型CSV文件。 “大”是指几千兆字节(因此一次将整个CSV读取到内存中是不切实际的)。 到目前为止,我已经尝试了以下选项: 使用TextIO.read(
我是一名优秀的程序员,十分优秀!