从发布/订阅到ElasticSearch的Apache光束流
从发布/订阅到ElasticSearch的Apache光束流是一个涉及到数据流处理和搜索引擎的技术链条。下面是对这个问答内容的完善和全面的答案：
- 发布/订阅（Publish/Subscribe）是一种消息传递模式，用于在分布式系统中实现解耦和异步通信。发布者将消息发送到一个或多个主题（Topic），而订阅者则通过订阅这些主题来接收消息。这种模式可以实现高度可扩展的系统架构。
- Apache光束（Apache Beam）是一个用于大规模数据处理的开源分布式编程模型和SDK。它提供了一种统一的编程接口，使得开发人员可以在不同的数据处理引擎上编写一次代码，例如Apache Flink、Apache Spark、Google Cloud Dataflow等。光束流支持批处理和流处理，并且具有高度的可扩展性和容错性。
- ElasticSearch是一个开源的分布式搜索和分析引擎，基于Apache Lucene构建。它提供了强大的全文搜索、结构化查询、实时数据分析和可视化等功能。ElasticSearch使用倒排索引来加速搜索，支持水平扩展和高可用性。
- 将发布/订阅与ElasticSearch结合使用，可以实现实时数据流处理和搜索分析的场景。通过使用Apache光束流，可以将数据从发布者传输到订阅者，并进行必要的数据转换和处理。而ElasticSearch则可以用于存储和索引这些数据，以便进行快速的搜索和分析。
推荐的腾讯云相关产品和产品介绍链接地址：
- 腾讯云消息队列 CMQ（https://cloud.tencent.com/product/cmq）：提供高可靠、高可用的消息队列服务，可用于实现发布/订阅模式。
- 腾讯云流计算 Oceanus（https://cloud.tencent.com/product/oceanus）：提供实时数据处理和分析的托管服务，可用于处理数据流并将其发送到ElasticSearch。
- 腾讯云ElasticSearch（https://cloud.tencent.com/product/es）：提供稳定可靠的ElasticSearch托管服务，支持高性能搜索和分析。
请注意，以上推荐的腾讯云产品仅供参考，具体选择应根据实际需求和情况进行评估。
相关·内容
1回答
从发布/订阅到ElasticSearch的Apache光束流
java、elasticsearch、google-cloud-dataflow、apache-beam、elasticsearch-rest-client
目前，我使用的是direct runner，但计划在Google Cloud Dataflow上部署该解决方案。然后，当我尝试使用Beam的ElasticSearch连接器执行写操作时，我开始遇到一些错误。我所做的基本上是这样的：
浏览 31提问于2019-11-21得票数 0
回答已采纳
1回答
如何使用BigQuery存储读取API定义Apache Beam中的最大流数
google-cloud-platform、google-bigquery、google-cloud-dataflow、apache-beam
我想从BigQuery中读取数据并将其发布到Pubsub。为此，我想使用Apache光束，并在数据流中运行此管道。而且我的BQ查询需要很长时间才能执行。因此，希望使用BigQuery的Storage Read API来减少这一时间，该API支持使用流读取数据。尽管Beam支持使用存储API，但我找不到设置最大流数的文档。谁能帮助我如何设置从BigQuery读取数据的Apache<e
浏览 25提问于2021-05-02得票数 0
1回答
PubSub确认截止日期
google-cloud-platform、google-cloud-pubsub、google-cloud-run
我有一个云函数，它将消息发布到PubSub，并触发云运行来执行归档文件过程。当有大文件时，我的cloud Run python代码需要一些时间来处理数据，它看起来像是PubSub在20秒(默认确认截止时间)后重试消息，这将从我的Cloud run触发另一个实例。云函数发布消息代码： # Publishes a message publish_future = publisher.publish(topic_path, data=message_bytesexcept Excep
浏览 17提问于2021-07-02得票数 2
回答已采纳
1回答
Apache / PubSub处理文件前的时间延迟
google-cloud-platform、google-cloud-dataflow、apache-beam、google-cloud-pubsub
我需要延迟处理或发布文件名(文件)。我在找最好的选择。1)延迟发布消息，以便立即处理它，但要在好的/预期的时刻进行处理
2)延迟处理检索<
浏览 0提问于2018-12-21得票数 0
1回答
在Apache Beam Python SDK中发布/订阅到数据存储批处理数据流作业可能吗？
python-2.7、google-cloud-dataflow、google-cloud-pubsub、apache-beam-io
我有一个发布/订阅主题，它将定期(通常每隔几天或几周一次，但有时更频繁)接收批量消息。我想启动一个批处理数据流作业来读取这些消息，执行一些转换，将结果写入数据存储，然后停止运行。当一批新的消息发出时，我想开始一项新的工作。我已经阅读了Apache Beam Python SDK文档和许多问题，但对一些事情仍然不确定。
是否可以将发布/订阅IO作为非流作业的一部分进行读取？然后，同一作业是否可以使用
浏览 2提问于2019-02-21得票数 1
2回答
Google Cloud Dataflow - Apache光束-管道关闭钩子
google-cloud-dataflow、apache-beam
想知道是否有某种“钩子”来放置一段代码，这些代码将在apache beam管道关闭时执行(无论出于什么原因--崩溃，取消)
每次数据流停止时，我都需要删除pubsub主题上的订阅。
浏览 12提问于2017-08-17得票数 1
2回答
google-cloud-dataflow vs apache-apache
google-cloud-dataflow、apache-beam
令人困惑的是，每个关于数据流的谷歌文档都说它现在是基于Apache光束的，并将我引导到光束网站。此外，如果我查找github项目，我会发现google dataflow项目是空的，并且所有的项目都转到apache see repo。假设我现在需要创建一个管道，根据我从Apache光束中读到的，我会这样做：from apache_
浏览 1提问于2017-06-16得票数 3
1回答
Apache Beam如何管理动态检查点？
apache-beam、amazon-kinesis、apache-beam-io、amazon-kcl
我有一个在Apache光束中开发的流管道(使用Spark Runner)，它从kinesis流中读取。我正在寻找Apache光束中的选项来管理运动检查点(即定期存储运动流的当前位置)，以便它允许系统从故障中恢复并在流中断的地方继续处理。有没有一个条款可用于支持类似于Spark Streaming的运动检查点(参考链接- )？
浏览 9提问于2020-06-08得票数 1
1回答
Apache光束从发布/订阅JSON获取CSV文件
apache-beam
我已经将csv文件推送到Google Storage，并订阅了pubsub，当它们到达时会通知我。我正在尝试完成的是编写一个beam程序，该程序将从pubsub订阅中获取JSON数据，解析出文件位置，然后从GS读取csv文件，然后对这些文件进行处理。到目前为止，我有这样的想法：
Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class
浏览 0提问于2017-09-27得票数 0
2回答
如何从Map<Value，Value>获取值？
java、google-cloud-platform、apache-beam、msgpack、apache-beam-io
在Apache光束转换中，我可以成功地将从Google Cloud发布/订阅中读取的发布/订阅值(以MessagePack格式)转换为MessagePack Value对象的映射，如下所示：
@我是否需要以不同的方式转换值，或者需要以不同的方式检索它们？
浏览 4提问于2018-11-08得票数 1
2回答
卡夫卡、菲比和洛什之间有什么区别？
apache-kafka、logstash、filebeat
我在一些小项目中使用了这3种技术，但我有一种感觉，他们的kindah做同样的工作，其中三人从一个地方接收数据并发送到另一个地方(他们负责数据摄取)，我错了吗？考虑到吞吐量，我什么时候应该选择卡夫卡或文件拍来发送数据呢？
浏览 7提问于2021-02-18得票数 0
1回答
通过数据流SQL实现云发布/订阅到BigQuery
google-bigquery、google-cloud-dataflow、gcloud、google-cloud-pubsub
我想了解数据流管道的工作原理。 在我的例子中，我有一些定期发布到云发布/订阅的内容，然后Dataflow将其写入BigQuery。通过的消息量数以千计，所以我的发布者客户端有1000条消息的批处理设置，1mb和10秒的延迟。问题是，在批处理中发布时，Dataflow SQL是接收批处理中的所有消息并将其一次性写入BigQuery，还是一
浏览 35提问于2021-11-23得票数 0
回答已采纳
2回答
Flink将Apache光束捆绑在一起，不显示单独的任务块
apache-flink、apache-beam
我有一个运行在Flink runner上的光束管道，如下所示： ? 光束中的整个管道都被捆绑到这个区块中。为什么Flink用这种方式来说明它？ ? 当然，管道步骤中的不同pardo应该由不同的块和工作者来表示(就像在Google的数据流中一样)。是否需要在Flink运行器中设置某些内容以显示更细粒度的任务，或者Flink UI是否仅为原生Flink代码构建，而这正是该UI处理Apache</em
浏览 24提问于2021-07-05得票数 1
1回答
Dataflow中的自定义Apache Beam Python版本
python、google-cloud-dataflow、apache-beam
我想知道是否有可能有一个自定义的阿帕奇光束Python版本运行在谷歌数据流。在公共存储库中不可用的版本(在撰写本文时: 0.6.0和2.0.0)。例如，来自Apache Beam官方存储库的HEAD版本，或与此相关的特定标签。但我还没有设法得到当前<em
浏览 16提问于2017-07-27得票数 4
回答已采纳
1回答
IIDR CDC Kafka在AWS上
amazon-web-services、apache-kafka、cdc、ibm-infosphere、ibm-data-replication
我们试图在AWS上将db2-IIDR (IBM )的数据发布到Kafka。由于以下错误，订阅失败-
An error occurred during the conversation with Kafka.Error: org.apache.kafka.common.errors.TimeoutException: Batch containing 1 record(s) expired due to timeoutwhile requesting metadata from brokers for c
浏览 1提问于2019-01-25得票数 0
回答已采纳
1回答
作为云监视日志用户的
amazon-web-services、amazon-kinesis、amazon-kinesis-firehose、amazon-cloudwatchlogs
我的要求是从AWS收集VPC流日志。我已经创建了一个VPC流日志，并将其与云监视日志组“流日志组”相关联。从CloudWatch日志订阅中，我想通过AWS来订阅它，但是在订阅列表中我找不到"Kinesis“，只有"Lambda”和"ElasticSearch服务“。你能帮我一下吗。我们不能订阅CloudWatch日志使用AWS动态消防软管。
浏览 0提问于2017-01-12得票数 4
1回答
有没有办法从BigQuery SDK / API获取实时数据？
google-bigquery
BigQuery数据集中获取实时数据，所以我努力研究了解web套接字，如pusher和laravel-websocket，我还对它们进行了测试 但是有一件事我不明白，如果我把Laravel SDK和我的BigQuery项目连接起来，我没有发现BigQuery提供给我们的任何事件或方法，因为在我的场景中，大多数数据来自IoT设备，而设备使用bigQueryAPI来提供数据，我也希望这些设备的数据是实时的。然后，我发现有一种方法可以将BigQuery dataset与firebase连接起来，然后每当数据集中
浏览 20提问于2021-07-07得票数 0
1回答
星星之火卡夫卡连接与麋鹿堆栈
apache-spark、elasticsearch、apache-kafka、logstash、spark-streaming
我想使用kafka连接和火花流插入elasticsearch，然后可视化地使用kibana进行BI用例，请您帮助我解决我的问题--我不想从哪里开始--是否有任何使用这些技术的项目可以帮助我了解实现的逻辑
浏览 2提问于2017-03-25得票数 0
回答已采纳
1回答
使用Lambda与带有CloudWatch订阅过滤器的Firehose发布到ElasticSearch
elasticsearch、amazon-cloudwatch、amazon-kinesis-firehose
最终，我们希望处理日志以从查询字符串和响应头中提取统计信息，并将结果发布到ElasticSearch或S3。在Lambda中处理数据并将其发送到ElasticSearch/S3。
创建CloudWatch订阅筛选器并从Firehose目标进行订阅。使用Lambda转换函数提取数据，将其放回流中，并让Firehose将数据发布到ElasticSearch/S3。订阅筛选器基本上会传
浏览 2提问于2018-07-07得票数 0
2回答
如何以编程方式取消运行时间过长的数据流作业？
google-cloud-dataflow、apache-beam
我通过Python API在Dataflow上使用Apache光束从Bigquery读取数据，对其进行处理，然后将其转储到Datastore接收器中。不幸的是，作业经常会无限期地挂起，我不得不手动停止它。当数据被写入到Datastore和Redis中时，从数据流图中，我注意到只有几个条目被卡住，作业被挂起。因此，当具有15台16核计算机的作业运行9小时(正常情况下，该作业运行30分钟)时，会导致巨大的</em
浏览 3提问于2018-06-25得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
活动推荐
腾讯云开发者
