Flink可以接收http请求作为数据源吗？
Flink是一个开源的流处理框架，它主要用于处理实时数据流。它可以接收各种类型的数据源作为输入，包括http请求。
在Flink中，可以通过使用Flink的Source函数来定义数据源。对于http请求作为数据源，可以使用Flink提供的HTTPSource函数来实现。HTTPSource函数可以监听指定的http端口，并将接收到的http请求转化为Flink的数据流。
使用Flink的HTTPSource函数作为数据源有以下优势：
- 灵活性：通过接收http请求作为数据源，可以方便地从各种来源获取数据，如Web应用程序、移动应用程序等。
- 实时性：Flink的流处理能力可以实时处理接收到的http请求数据，使得数据处理结果能够及时反馈给用户。
- 可扩展性：Flink支持水平扩展，可以通过增加更多的任务节点来处理大规模的http请求数据。
应用场景：
- 实时日志分析：通过接收http请求作为数据源，可以实时地分析和处理日志数据，如异常检测、用户行为分析等。
- 实时监控：可以通过接收http请求作为数据源，实时地监控系统的状态和性能指标，如服务器负载、网络流量等。
- 实时推荐系统：通过接收http请求作为数据源，可以实时地处理用户的请求，并给出个性化的推荐结果。
腾讯云相关产品推荐：
腾讯云提供了一系列与流处理相关的产品和服务，可以与Flink结合使用，以构建完整的流处理解决方案。以下是一些推荐的腾讯云产品和产品介绍链接地址：
- 云原生应用引擎（Cloud Native Application Engine）：提供了基于容器的应用托管服务，可以方便地部署和管理Flink应用程序。详细介绍请参考：https://cloud.tencent.com/product/tke
- 云数据库TDSQL（TencentDB for TDSQL）：提供了高性能、可扩展的关系型数据库服务，可以作为Flink应用程序的数据存储。详细介绍请参考：https://cloud.tencent.com/product/tdsql
- 云服务器（Cloud Virtual Machine）：提供了弹性、安全的云服务器实例，可以用于部署Flink集群。详细介绍请参考：https://cloud.tencent.com/product/cvm
请注意，以上推荐的产品和服务仅供参考，具体选择应根据实际需求和情况进行。
相关·内容
作为一个玩具示例，假设我希望构建一个web应用程序，该应用程序接收带有ID和$ payed的HTTP请求，并以ID在24小时窗口中支付的总额返回响应。使用Apache，如果输入和输出是Kafka流，当然可以创建该应用程序。
，但是否支持/可能创建一个Flink应用程序，其中输入是HTTP请求，输出是对HTTP请求的响应？
如果Flink不支持这一点，是否还会有另一个数据流框架支持这一点？
(对于上面的玩具示例，我想有比使用Flink更简单的解决方案，但我的实际使用是通过HTTP“流”请求的许多多个窗口和其他有状态的计算。)
浏览 4提问于2020-06-16得票数 0
回答已采纳
我使用elasticsearch 6.2.2作为flink中的接收器，但在阅读apache flink文档时，我可以看到用于5.x版的连接器，但没有提到6.x版的连接器。
flink是否支持elasticsearch 6.x的任何连接器？或者，我是否应该创建任何http客户端，并在每次创建或更新新文档时发出post请求？或者他们有其他的技术？
浏览 14提问于2018-02-28得票数 0
我正在对以下代码进行一些查询：
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
DataStream<Row> ds = SourceHelp.builder().env(env).consumer010(MyKafka.builder().build().kafkaWithWaterMark2())
.rowTypeInfo(MyRowType.builder().build().typeInfo())
.build().source4()
浏览 1提问于2018-09-17得票数 1
我们一直在和弗林克玩。到目前为止，我们已经在Hadoop2.x/纱线上使用了Spark和标准M/R。
除了纱线上的Flink执行模型，AFAIK不是动态的火花，执行者在纱线中动态地获取和释放虚拟芯，问题的要点如下。
Flink看起来真是太棒了：用于流媒体API的，我只想说它是非常出色的，而且是过头的。
批处理API的：处理图非常强大，并且以一种独特的方式被优化和并行运行，它比Spark和其他方面更多地利用了集群可伸缩性，并对共享共同处理步骤的非常复杂的进行了优化。
我发现的唯一缺点，我希望只是我的误解和缺乏知识，是它似乎不喜欢数据-本地处理时，规划使用HDFS输入的批处理作业。
不幸的是，这不
浏览 0提问于2016-07-30得票数 4
回答已采纳
1回答
我正在使用Flink 1.11使用，但在启动应用程序时出现了以下错误：
java.util.concurrent.TimeoutException: The heartbeat of TaskManager with id 421563c271e57acb4592f9d447d45b42 timed out.
at org.apache.flink.runtime.resourcemanager.ResourceManager$TaskManagerHeartbeatListener.notifyHeartbeatTimeout(ResourceManager.java:1202)
浏览 4提问于2021-12-28得票数 0
2回答
异步闪烁计时器
、、
我有一个关于Flink和它的计时器服务的问题。
我有一个使用计时器的keyBy流，当计时器被调用时，我需要发送一个http请求，这可能需要一段时间来响应。我的问题是，我应该进行异步http调用吗？或者flink已经将计时器调用作为一个新线程，每个键都是异步的？
提前感谢
浏览 21提问于2020-02-17得票数 1
我集成了两个工具Apache NiFi和Apache Flink。NiFi获取事件并将它们发送到Flink，之后Flink将这些事件经过一些处理返回给相同的NiFi。我在Flink中构建了Nifi的源代码和接收器。整个过程可以工作，但是接收器的性能非常差(大约每秒10个事件)。
如果我移除接收器(仅打印输出)，处理速度会快得多。
我想我可以使用setParallelism()来改变接收器进程的并行性，这当然会有帮助，但是基本吞吐量太低了。我也尝试过使用requestBatchCount(1000)，但是什么都没有改变。
可能我的问题与事务有关。在每个事件接收器之后，等待关闭事务，但我不确定，也
浏览 3提问于2018-07-24得票数 1
我知道，通过设计和开箱即用的请求和回复数据处理是不可能的Flink。但是考虑一下，例如一个遗留的TCP应用程序，它打开到服务器的连接，并期望在tha相同的连接中得到响应。
例如，考虑一个遗留应用程序，其中客户端通过TCP和自定义协议连接到服务器。他们发送一些状态信息，并期望命令作为响应，其中命令可能依赖于当前状态。
是否有可能构建一个将TCP消息输入到处理中的组合源，以及接收处理结果的接收器？
构建一个源，它接受TCP连接并从消息中创建事件看起来很简单，但是在同一个工作者(！)上获得对corrent接收器的相应响应。要将响应发送到客户端，请参阅棘手的内容。
我知道，这可以通过外部组件来实现，但
浏览 0提问于2019-08-26得票数 0
回答已采纳
从以前的来看，Flink似乎实现了精确的一次
在成功的预提交之后，必须保证提交最终成功。
我认为“成功的预提交”是由Flink任务管理器实现的，而“最终成功”是由Flink接收器实现的。
Flink接收器节点如何实现“最终成功”？
这一次的机制与检查点有关吗？
浏览 1提问于2019-12-23得票数 1
我有一个Flink应用程序，它使用带有多个分区的Kafka主题上的传入消息，然后进行一些处理，然后将它们发送到通过HTTP发送到外部服务的接收器。有时下游服务正在下降，流处理需要停止，直到它重新开始工作为止。
我正在考虑两种方法。
当Http接收器无法发送输出消息时，抛出一个异常。这将导致根据配置的重新启动策略重新启动任务和作业。最终，下游服务将恢复，系统将继续在中断的地方运行。让Sink睡眠并在失败时重新尝试；它可以持续地这样做，直到下游服务恢复为止。
根据我的理解和我的PoC，我将完全失去-至少一次保证，因为接收器本身是外部状态。据我所见，您不能进行简单的HTTP端点事务处理，因为它需要实
浏览 2提问于2021-11-21得票数 0
回答已采纳
运行Apache Flink 1.8.0。我可以在http://localhost:8081上访问工作的UI。我还验证了我的工作可以在集成开发环境中工作，使用./flink run通过命令行提交。我甚至已经通过UI上传了作业。这项工作做得很好。 但当我查看UI上的统计数据时，我只看到一个微调器。例如:字节接收，接收的记录，发送的字节，发送的记录都有一个微调。 我也在我的RichAsyncFunction里做了 UI的文档中并不清楚，我们是否需要在Flink服务器上启用某些功能？ private transient Counter lookupCounter;
@Override
public
浏览 32提问于2019-05-04得票数 1
1回答
我正在学习如何使用Flink处理流数据。
我已经成功地编写了一个示例，它是从数据源接收和反序列化流数据，转换它并打印输出。
现在我在考虑如何处理Flink中的OOM例外。
例如，如果存在背压问题，这意味着如果从数据源发送数据的速度比Flink的操作符中处理数据的速度快，我的理解是，RAM将在一段时间内耗尽。如果这个案子发生了呢？如何处理这种异常？是否有可能忽略某些输入，从而使进程不会导致任何错误？
换句话说，我期待一些机制如下：
if (RAM is almost exhausted)
ignore the coming data
else
process the coming
浏览 4提问于2020-06-08得票数 0
回答已采纳
1回答
我试着用阿帕奇光束和Flink在AWS KDA。这个管道读取一些动态数据，经过简单的转换后，它尝试使用以下策略将结果分组为一个碎片：
.apply("Creating micro-batches", Window.<XXX>into(new GlobalWindows())
.triggering(Repeatedly.forever(AfterFirst.of(AfterPane.elementCountAtLeast(windowSizePanes),
浏览 3提问于2021-12-20得票数 0
回答已采纳
1回答
我们有AsyncFunction，异步操作是使用完成的
class Foo[A,B] extends AsyncFunction[A, B] with {
val akkaConfig = ConfigFactory.load()
implicit lazy val executor: ExecutionContext = ExecutionContext.fromExecutor(Executors.directExecutor())
implicit lazy val system = ActorSystem("MyActorSystem", akkaConfi
浏览 0提问于2019-07-04得票数 0
回答已采纳
在使用java REST客户端将jar上传到Flink 1.5.0服务器时，我们得到了错误响应{“error”：“Internal server error: null"}。同样的代码在Flink 1.4.2中也能正常工作。事实上，我们可以看到jar是从Flink GUI上传的。但是错误的状态中断逻辑。有什么建议吗？
HttpPost uploadFile = new HttpPost(flinkJobManagerUrl + "/jars/upload");
MultipartEntityBuilder builder = Multip
浏览 172提问于2018-06-01得票数 1
回答已采纳
我试图理解Apache仪表板显示的"Bytes接收/发送“意味着什么。对于某些上下文，CSV文件托管在HDFS服务器上，我正在将结果写入我的计算机上的本地TXT文件。Flink也在我的机器上本地运行。考虑到这一点，"Bytes server“似乎意味着"Bytes从HDFS服务器发送到我的机器”，而"Bytes接收“似乎意味着"Bytes从我的机器发送到HDFS服务器”。，这是正确的解释吗？
时间线显示的重叠任务也让我有点困惑。奇怪的是，连接在过滤前两个数据集完成之前就开始了。，这是预期的行为吗?如果是的话，为什么？
下面是我关于正在
浏览 1提问于2019-05-19得票数 1
回答已采纳
我正在构建一个应用程序，它需要聚合来自部署在不同区域的一系列传感器的度量。这些措施是使用Kafka摄取的。我是Flink的新手，但我已经知道如何使用窗口聚合事件并将它们发送到接收器中。但是，我还需要将每个区域的聚合值与来自外部数据库(在我的例子中为Postgres)的阈值(也是每个区域)进行比较。这些阈值也可以随时间更新或在创建新区域时添加。有什么建议吗？谢谢你，尤克斯
浏览 10提问于2021-08-10得票数 0
回答已采纳
我想从一个url中获得JSON数组形式的输入流。如何设置源，以便在apache flink中使用datastream.In short连续获得输入我希望从url连续获得json数据，而不关闭flink作业。
浏览 3提问于2018-03-08得票数 1
我已经开发了一个Flink程序，可以读取Twitter上的推文，并将它们推送到Kafka上。然后，它从Kafka获取tweet并对其进行处理。
"Tweets“转换从tweet的文本中提取标签和用户，并在默认输出中发出它们，并在副输出中发出它们的每一对。
附加的图像是从Flink Web UI中拾取的。我不明白为什么Kafka源和Tweets处理操作符合并到一个任务中，主要是我希望Tweets接收器接收所有来自Kafka源的原始Tweet，而不是tweets处理操作符的输出。
程序是正确的吗？
这是代码的相关部分：
FlinkKafkaConsumer010<Strin
浏览 1提问于2018-06-30得票数 1
首先，我是Flink的一个完全的新手。我已经在Windows上安装了Apache。我开始用开始-集群。蝙蝠。印出来
使用一个JobManager进程和一个TaskManager进程启动本地集群。您可以通过生成的shell窗口中的CTRL终止进程。默认情况下，在上使用Web接口。
不管怎么说，当我提交这份工作的时候，我有一堆信息：
调试org.apache.flink.runtime.rest.RestClient -接收到的响应{“状态”：{“id”：“IN_PROGRESS”}}。
在的web日志中，我看到：
2019年-02-15 16:04:23.571 flink-a
浏览 1提问于2019-02-15得票数 1
1回答
我正在使用一个名为Localstack的产品在本地模拟亚马逊S3，它是一个Flink作业的流文件接收器。 在运行日志中，我可以看到Flink忽略了本地堆栈，并试图联系亚马逊S3。 Received error response: org.apache.flink.fs.s3base.shaded.com.amazonaws.services.s3.model.AmazonS3Exception: Service Unavailable
Retrying Request: HEAD https://s3.amazonaws.com /testBucket/ 在flink-conf.yaml中
浏览 70提问于2020-08-11得票数 2
1回答
我有两个来自两个Kafka主题的流表，我想加入这些流并对所连接的数据执行聚合功能。需要使用滑动窗口连接流。在连接和窗口数据时，我将得到一个错误Rowtime attributes must not be in the input rows of a regular join. As a workaround you can cast the time attributes of input tables to TIMESTAMP before.。
下面是代码片段
select cep.payload['id'] , ep.payload['id'] ,
浏览 25提问于2022-05-24得票数 0
1回答
Apache通过从检查点恢复作业来保证在故障和恢复时处理一次，检查点是分布式数据流和操作符状态的一致快照(Chandy-Lamport算法用于分布式快照)。此保证在故障转移时恰好是一次。
在正常的集群操作中，Flink如何保证只处理一次，例如给定从外部源读取的Flink源(例如Kafka)，Flink如何保证事件从源读取一次？事件源和Flink源之间是否存在某种类型的应用程序级别？另外，Flink如何保证事件准确地从上游运营商到下游运营商之间只发生一次？对于接收到的事件，这也需要任何形式的附加条件吗？
浏览 3提问于2020-09-30得票数 3
回答已采纳
我想使用NAT流服务器来流式传输数据，并使用Flink来处理数据。
如何通过NATS流服务器使用apache flink处理实时流数据？
浏览 8提问于2020-03-30得票数 2
1回答
我要把唱片从蜂箱搬到卡夫卡。我找不到直接向flink数据集添加kafka接收器的方法。因此，我使用了一个解决方案，在flink数据集上调用映射转换，在map函数中对给定的记录使用kafkaProducer.send()命令。
我面临的问题是，我无法在每个工作节点上执行kafkaProducer.flush()，因此用kafka编写的记录数量总是略少于dataset中的记录数。
有什么好办法来处理这件事吗？我可以在flink中添加一个kafka接收器到dataset吗？还是将kafkaProducer.flush()作为终结器调用的方法？
浏览 1提问于2018-09-13得票数 3
回答已采纳
我是Apache flink的新手，正在构建一个简单的应用程序，其中我从kinesis流中读取事件，例如
TestEvent{
String id,
DateTime created_at,
Long amount
}
对由id键控的上述流的字段amount执行聚合(sum)。该转换相当于SQL select sum(amount) from testevents group by id，其中testevents是到目前为止接收到的所有事件。聚合结果以flink状态存储，我希望通过API公开结果。有没有办法做到这一点？
PS:我们可以将闪烁状态存储在dynamoDB中并在那里创建一个应
浏览 2提问于2019-11-25得票数 0
我正在跟踪Flink 。打开Flink SQL客户端看起来很好，但是SELECT 'hello world'不能工作。
$ /data/apache/flink/flink-1.12.0/bin/sql-client.sh embedded
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/data/apache/flink/flink-1.12.0/lib/log4j-slf4j-impl-2.12.1.jar!/org/slf4j/impl/Stati
浏览 14提问于2021-03-26得票数 1
回答已采纳
1回答
我想知道是否有可能同时使用Flink实现端到端的精确语义？
我的数据管道如下所示。
来源: kafka过程: flink接收器: kafka
来源: kafka进程: flink接收器:数据库(mysql或redis)
期待着你的回答。
谢谢。
浏览 2提问于2022-09-26得票数 -1
我有一个Flink作业，它的接收器是将数据写入MongoDB。接收器是RichSinkFunction的实现。
启用外部检查点。区间为5000轧机，方案为EXACTLY_ONCE。
Flink版本1.3，
Kafka (源主题) 0.9.0
我不能升级到Flink 1.4的TwoPhaseCommitSink。
我不怀疑
接收器在什么时候确认检查点屏障，在调用函数开始时还是在调用完成时？意味着它在确认障碍之前等待持久化(保存在MongoDB中)响应？
如果提交检查点是由异步线程完成的，那么在作业失败的情况下，Flink如何准确地保证一次呢？如果数据被接收器保存到Mong
浏览 3提问于2018-05-31得票数 1
1回答
闪存程序的执行必须被触发，例如使用execute()。否则Flink只会创建一个新的执行计划，对吗？我的问题是:在不触发执行的情况下处理延迟操作时，Flink的哪些组件被激活？
根据，有一个优化器负责构建数据流图。是否有更多的过程涉及？
是否有一种方法可以找到优化程序进程的id以便监视它？
浏览 2提问于2015-12-22得票数 1
回答已采纳
我正在尝试一个简单的Flink程序，它只接受一个文件，反转文件中的字符串并写出它。
程序正常运行，只有个别的线路出现故障。
例如。
文件输入
Thing,Name
Person,Vineet
Fish,Karp
Dog,Fido
输出文件
Fish,praK
Thing,emaN
Person,teeniV
Dog,odiF
我原以为：
Thing,emaN
Person,teeniV
Fish,praK
Dog,odiF
下面是我为实现这一目标而编写的程序：
package testflink;
import java.util.Iterator;
import java.util.Str
浏览 1提问于2017-06-30得票数 0
回答已采纳
1回答
接收器的invoke方法似乎没有办法做异步？例如返回Future
例如，redis连接器使用jedis同步执行redis命令：
然后它会阻塞flink的任务线程，等待来自redis服务器的网络响应？！其他操作符是否有可能在同一个线程中与接收器运行？如果是这样的话，它也会阻止他们吗？
我知道flink有异步api，但它似乎不适用于接收器impl？
浏览 6提问于2017-12-08得票数 2
回答已采纳
我确实从flink 1.11升级到1.15.1在S3上编写S3文件。由于connectors.fs现在已经过时，我必须更改我以前使用过的接收器: EventuallyConsistentBucketingSink：
https://github.com/sjwiesman/flink/blob/83a6400e2587b067d08a64bc7e10edd4b57e71b4/flink-connectors/flink-connector-filesystem/src/main/java/org/apache/flink/streaming/connectors/fs/consistent/E
浏览 5提问于2022-07-19得票数 1
我正在尝试写一个Flink应用程序，它将从Kafka读取数据，并通过自定义Hbase接收器在Hbase中增加计数。我正在研究容错，我看到Flink将从最新检查点的偏移量重新启动作业。这不是意味着我的Hbase接收器将产生重复项吗？我是否必须实施自己的重复数据消除逻辑？
浏览 5提问于2019-09-10得票数 0
我想设置Flink，这样它就可以将数据流从Apache转换并重定向到MongoDB。出于测试目的，我构建在flink流-Connectors.kafka示例()之上。
卡夫卡流是正确红色的链接，我可以映射他们等，但问题发生时，我想保存每一个接收和转换的信息到MongoDB。我发现的关于MongoDB集成的唯一例子是来自github的flink-mongodb测试。不幸的是，它使用的是静态数据源(数据库)，而不是数据流。
我认为应该有一些DataStream.addSink实现的MongoDB，但显然没有。
实现这一目标的最佳途径是什么？我是否需要编写自定义接收器函数，还是可能遗漏了什么？也许应
浏览 3提问于2016-02-02得票数 11
回答已采纳
我使用flink来处理大量的数据并添加到HBase.To中，提高了hbase的接收速度，我在flink接收器operate.When上缓冲了一些数据，达到了批处理阈值，数据将发送到HBase。
但是，如果在到达批处理阈值之前接收器操作失败，缓冲区上的数据会丢失吗?如果我一个一个地向HBase发送数据，速度似乎太慢了。
有没有人对如何最好地解决这一问题有任何建议？
浏览 1提问于2019-08-26得票数 0
回答已采纳
Flink在Flink应用程序(https://ci.apache.org/projects/flink/flink-docs-stable/dev/types_serialization.html)的飞行前阶段执行重要扫描。创建源、运算符和接收器的行为会导致Flink扫描在给定流的拓扑中使用的对象的数据类型，因为很明显，Flink将尝试基于此信息优化作业。 这种扫描是可配置的吗？我可以关闭它，强制Flink只使用Kryo序列化，而不需要或使用任何这些扫描的信息吗？ 我在一个专有的库中有一个非常大的，嵌套很深的类，它是自动生成的，当扫描它时，Flink似乎进入了一个非常大的无限循环，在运行几
浏览 18提问于2019-06-04得票数 0
我需要让一个AsyncIO富函数根据最新的规则集执行调用。对于像map这样的操作，我能够通过关注这篇博客文章来处理具有丰富功能的BroadcastConnectedStream：https://flink.apache.org/2019/06/26/broadcast-state.html 但是，创建AsyncIO函数需要DataStream作为输入，而BroadcastConnectedStream不需要(https://ci.apache.org/projects/flink/flink-docs-stable/api/java/index.html?org/apache/flink/s
浏览 29提问于2020-06-19得票数 0
回答已采纳
请您帮帮我--我正在尝试使用Apache Flink进行机器学习任务，外部集成/树形库(如XGBoost )，所以我的工作流程如下：
接收单个数据流，其中原子事件看起来像一个简单的向量event=(X1, X2, X3...Xn)，它可以想象为POJO字段，所以最初我们有DataStream<event> source=...
应用于同一事件源的许多特征提取代码：feature1 = source.map(X1...Xn)、feature2 = source.map(X1...Xn)等。为了简单起见，可以为所有特性提供DataStream<int> featur
浏览 2提问于2017-10-08得票数 1
回答已采纳
1回答
嗨，我想创建systemctl服务到tun flink本地服务器。
我可以直接从/bin/start-local.sh运行它。
但是当我使用这个服务定义时。
[Unit]
Description=Apache Flink
Documentation=http://flink.apache.org/
Requires=network.target remote-fs.target
After=network.target remote-fs.target
[Service]
Type=simple
User=beam1
Environment=JAVA_HOME=/usr/lib/jvm/ja
浏览 0提问于2017-10-23得票数 3
2回答
最近，我开始使用Apache，刚刚完成了我的第一个拓扑构建(都是用java编写的)。
作为下一步，我想把传感器值从一个TI SensorTag连接到一个树莓Pi，在这些拓扑之一。
我能够通过HTTP发送传感器数据，但我不确定如何实现一个工作喷口，它接收这些请求。
拓扑思想:它应该接收带有传感器值信息的HTTP请求，将这些数据发送到拓扑中，然后使用螺栓将它们写入文件/数据库中。
到目前为止，我在Stackoverflow上找到了一篇关于HTTP ()的文章，但遗憾的是，我没有被允许留下评论或者写任何私人消息(如果我错过了一些事情，很抱歉)。我不知道这个喷口到底是如何工作的，我想要一个示例代码。(
浏览 2提问于2016-11-25得票数 2
回答已采纳
问题: Flink应用程序无法接收和处理Kinesis connector在关闭(由于重新启动)时生成的事件
我们有以下Flink env设置
env.enableCheckpointing(1000ms);
env.setStateBackend(new RocksDBStateBackend("file:///<filelocation>", true));
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(pause);
env.getCheckpointConfig().setCheckpoin
浏览 0提问于2018-05-02得票数 0
1回答
我正在尝试使用aws的s3桶作为我的flink流的文件源。
因此，我需要设置一个IAM角色或AWS凭据。
不幸的是，我总是收到一个错误，它说他无法在获取安全细节。如果我用这个URL在flink worker上做了一个卷曲，我就会得到作为响应的rolename。当我将rolename添加到curl 时，可以从角色获得临时凭据。
因此，我的问题是:我怎样才能告诉弗林克，它应该发挥什么样的作用？我看不到任何属性可以告诉flink IAM角色的名称。还是我做错什么了？
在本地，它可以很好地设置aws凭据，但是我想用EC2实例的IAM角色来解决这个问题，因为它更漂亮。我在或中都找不到对这个过程的任何描述
浏览 5提问于2021-05-06得票数 0
我使用Elasticsearch连接器作为接收器将数据插入到Elasticsearch中(参见：https://ci.apache.org/projects/flink/flink-docs-release-1.7/dev/connectors/elasticsearch.html)。 但是，我没有找到任何连接器来从Elasticsearch获取数据作为源。 是否有任何连接器或示例使用Elasticsearch文档作为Flink pipline中的源？ 致以敬意， 阿里
浏览 24提问于2019-01-23得票数 1
我有一个flink流作业，它将处理的数据保存到elasticsearch版本8集群中，不幸的是，我使用的是flink版本1.13.6，它仍然使用elasticsearch版本7接收器。因此，在试图保存数据时，我会得到一个解析错误。
我有两个选项，要么删除当前集群并设置elasticsearch 7群集，要么为接收器启用兼容性模式。
我不能删除我的集群，因为我有大量的重要数据需要很长时间才能恢复，而且由于快照从高版本到低版本都不能工作，所以我还有第二个选择。
我在网上搜索发现：。
问题是我不知道该在哪里启用它。这是我的代码：
import java.io.Serializable;
import
浏览 6提问于2022-07-05得票数 0
回答已采纳
我在Apache Flink中关于多个流的迭代的问题。
我是Flink初学者，目前正在尝试在Flink上执行递归查询(例如，datalog)。
例如，查询每隔5分钟(翻滚窗口)计算一次传递闭包。如果我有一个输入流inputStream (由初始边缘信息组成)，另一个由inputStream初始化的outputStream (传递闭包)。我想通过加入outputStream来迭代地丰富inputStream。对于每个迭代，反馈应该是outputStream，并且迭代将持续到不能在outputStream上添加更多边。我的传递闭包的计算应该每隔5分钟定期触发一次。在迭代期间，inputStream
浏览 9提问于2018-02-13得票数 0
回答已采纳
在读取此之后，我正在尝试访问端点以启动一个Flink作业，如下所示-
curl -k -v -X POST -H "Content-Type: application/json" --data '
{
"programArgsList": [
"--runner",
"FlinkRunner",
"--inputTopicName",
"inputTopicNameValue",
"--Argum
浏览 28提问于2019-11-13得票数 1
回答已采纳
2回答
我的flink应用程序执行以下操作
资料来源:阅读卡夫卡记录形式的数据。
split:基于某些标准
窗口:10秒的时间窗口，将其聚合为一个大容量记录
接收器:将这些大容量记录转储到elasticsearch
我面临的问题是flink使用者无法保存数据10秒，并抛出以下异常：
由: java.util.concurrent.ExecutionException: java.io.IOException引起的:状态的大小大于允许的最大内存支持状态。Size=18340663，maxSize=5242880
我不能应用countWindow，因为如果记录的频率太慢，那么ela
浏览 3提问于2019-01-30得票数 2
我试图找到在Dataproc上运行Apache的正确方法，并使用Google /Sub作为源/接收器。当我创建Dataproc集群时，在将应用到最近的映像之后，Flink 1.6.4将被安装。
问题是，只能从Flink版本1.9.0开始使用。
那么，我的问题是，什么是正确的方式来使用所有这些一起？我应该用最新的Flink来建立自己的gce形象吗？有一个已经存在吗？
浏览 2提问于2019-09-18得票数 4
回答已采纳
我有几个Flink作业，它们接收来自一系列Kafka主题的数据，进行一些聚合，并将结果发布到Kafka主题中。
聚合部分在某种程度上是困难的。我必须从几个HTTP端点检索一些信息，并以特定的格式将响应放在一起。问题是有些出站HTTP调用有时会超时，所以我需要一种重试它们的方法。
我想知道是否有一种规范的方法在Flink操作符中完成这样的任务，而不需要完全手工地做一些事情。如果没有，建议采取什么办法？
浏览 7提问于2022-08-17得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
