我是新的Flink，并试图创建一个原型，可以做一个连续的查询以及按需的查询。第一部分已经起作用了(使用预定义的查询)，但与第二个部分斗争。有可以使用的SQL。但是，我们希望能够从web端点执行此操作。
我已经看到了Flink木星笔记本集成，这告诉我，这应该工作的方式从一个网页界面。但是，不知道如何做到这一点，每次都不提交一个完整的JAR。
到目前为止，我的研究已经使我广泛地找到了这两种资源：
有一项尚未落实的要求。见和
有一个开源的JDBC驱动程序(还没有尝试过)-- --不确定这是否得到了维护，并说“这个项目处于早期阶段”。
对我们还能探索什么有什么建议吗？
浏览 12提问于2022-01-15得票数 0
回答已采纳
如何使用独立应用程序(本地执行环境或迷你集群)实现flink保存点。我在flink-config.yaml文件中配置了savepoint目录，但不确定如何在关闭应用程序之前获取保存点，以及如何在重新启动应用程序后恢复？
是否有任何方法或必须先使用flink群集，然后再使用CLI。
感谢你的帮助。谢谢
浏览 0提问于2020-08-18得票数 0
我试图使用flink 1.15.1编写一个简单的表API S3流接收器(csv格式)，并面临以下例外：
Caused by: org.apache.flink.util.SerializedThrowable: S3RecoverableFsDataOutputStream cannot sync state to S3. Use persist() to create a persistent recoverable intermediate point.
at org.apache.flink.fs.s3.common.utils.RefCountedBufferingFileStream
浏览 20提问于2022-07-11得票数 0
回答已采纳
在eclipse中尝试执行基本的Flink程序时，由于.print()被datastream_name.print()调用来打印我的数据，我得到了错误。
使用Java8
错误
Exception in thread "main" java.lang.reflect.InaccessibleObjectException: Unable to make field private final byte[] java.lang.String.value accessible: module java.base does not "opens java.lang" t
浏览 15提问于2022-06-26得票数 0
1回答
在一个受保护的Hadoop集群中，我试图访问Flink am页面并从纱线中登录，并看到以下错误：User %remoteUser are not authorized to view application %appID
原因似乎是Flink侧的纱线缺乏ACL的支持。
代码如何工作
消息来自使用ApplicationACLsManager类的hadoop/ uses /server/AppBlock类。这个类执行检查并引用在RMAppManager中设置的app信息：
this.applicationACLsManager.addApplication(applicationId,
浏览 3提问于2018-06-14得票数 0
Flink在这方面做得很好，而且flink sql客户端看起来非常有前途。我正在尝试理解，当用于交互式查询时，是否有可能将治理工具与flink sql客户端集成。
浏览 36提问于2020-04-30得票数 1
回答已采纳
我正在使用从Kafka到HDFS的Flink bucketing接收器。Flink的版本是1.4.2。
我发现每次重新启动作业时都会丢失一些数据，即使是使用保存点也是如此。
我发现如果我设置写入器SequenceFile.CompressionType.RECORD而不是SequenceFile.CompressionType.BLOCK，这个问题就可以解决。当Flink尝试保存检查点时，有效长度与实际长度不同，实际长度应该包括压缩数据。
但是，如果我们由于磁盘使用而不能使用CompressionType.BLOCK，这可能是一个问题。如何在重新启动作业时使用数据块压缩防止数据丢失？
这是Fl
浏览 0提问于2019-04-24得票数 0
2回答
在使用Scala代码将数据从输出NiFi端口传输到Flink方面，我需要一些帮助。
我被.addSource()函数卡住了。它要求额外的参数(OUT)，但是当我提供它们时，我总是得到一个错误。Scala代码和错误消息如下所示。
package flinkTest
import java.nio.charset.{Charset, StandardCharsets}
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api
浏览 0提问于2017-07-17得票数 0
回答已采纳
我们可以看到，在将Flink度量暴露于外部系统(如Prometheus )时，Flink使用Pushgateway而不是Prometheus通常的拉模型进行一般度量集合。
@Override
public void report() {
try {
pushGateway.push(CollectorRegistry.defaultRegistry, jobName);
} catch (Exception e) {
log.warn("Failed to push metrics to PushGateway with jobName
浏览 2提问于2018-08-19得票数 0
回答已采纳
1回答
我当前的Flink应用程序
基于Flink的状态函数3.1.1，它读取Kafka的消息，处理该消息，然后接收到Kafka的母题
应用程序已经部署在K8s上，下面是指南，并且运行良好：
基于标准部署，我打开了
我的目标
我想自动放大/缩小有状态函数。我还想知道如何创建更多的备用职务经理。
我对医管局的观察
我尝试将kubernetes.jobmanager.replicas设置在flink-config ConfigMap中：
---
apiVersion: v1
kind: ConfigMap
metadata:
name: flink-config
labels:
浏览 54提问于2022-01-26得票数 0
有没有一个地方我们可以快速参考和查看Apache beam和运行时环境(Flink，Spark等)之间的版本兼容性。我正在以一种艰难的方式学习它：我不确定我的问题有多少是由版本错误造成的。
目前我使用的是光束2.4.0，看看flink runner pom ()，它使用的是flink 1.4.0 jars。所以我假设Apache beam 2.4.0可以与Apache Flink 1.4.0一起工作。这是正确的解释吗？
不幸的是，Flink 1.4.0 ()没有官方的docker镜像
浏览 0提问于2018-06-14得票数 0
1回答
我有一个用Flink编写的流处理应用程序&我想从状态后端使用它的内部键值存储来计算流聚合。因为我要处理大量的聚合，所以我希望避免像当前提供的内存备份和文件备份实现那样，在Flink应用程序内部将它们维护在堆上。取而代之的是，我希望在Apache Ignite中维护一个状态缓存，这样就可以使用直写和直读特性在HBase中提供更可靠的备份。
理想情况下，我应该在每个物理节点上都有一个本地Ignite缓存，用于处理该节点上所有长期运行的Flink操作符的状态。例如，每个节点在一个8 GB的容器中有一个单独的Ignite节点可用，无论它是运行1个还是10个Flink运算符。
问题是我希望Fli
浏览 3提问于2016-02-08得票数 1
我目前的尝试是：
Eclipse 4.4.2露娜使用scala-ide插件
为Scala2.11重新构建Flink (默认情况下，maven下载官方Flink build，也就是Scala2.10。这给IDE带来了一些问题，例如可怕的宏不兼容。
但是，这样的工作有点麻烦，因为你需要：
每次更新到最新源代码时，都要重新构建Scala2.11的Flink。
拥有正确的maven设置，以便在Eclipse中构建项目时，您的项目依赖关系不会被Flink正式构建(即用scala 2.11 Flink build覆盖scala 2.10 build)覆盖。
浏览 1提问于2015-09-15得票数 1
回答已采纳
1回答
我试图使用Flink REST通过管道自动化Flink作业提交过程。要调用任何Flink Rest端点，我们应该了解接口IP。对于我的POC，在CLI上运行flink-session命令之后，我得到了IP，但是从代码中获得它的方法是什么呢？
对于自动化，我计划按顺序调用下面的REST
请求。获取(‘) //获取运行的作业Id
requests.post(') //取消保存点作业
requests.get(' :savepointId') //获取保存点状态
请求。发布(“”). //上传jar用于新作业
request
浏览 9提问于2022-10-12得票数 0
我正在尝试理解在zookeeper升级期间flink作业管理器的行为是否是预期的。 我在kubernetes中运行flink 1.11.2，使用zookeeper服务器3.5.4-beta。当我进行zookeeper升级时，zookeeper有20秒的停机时间。我希望flink作业在这20秒内重新启动，或者在日志中显示一些警告。相反，我看到整个flink JVM崩溃(后来pod重启)。 我以为flink会在内部重试zookeeper请求，所以我很惊讶它崩溃了。这是意料之中的，还是说这是一个bug？ 从日志中 org.apache.flink.shaded.zookeeper3.org.apa
浏览 75提问于2021-02-09得票数 0
回答已采纳
我曾计划用python statefun示例(链接：https://github.com/apache/flink-statefun/tree/master/statefun-examples/statefun-python-greeter-example)来交流嵌入式有状态函数 但是(总会有一个但是)我一直在努力构建python发行版。我想我的配置是正确的。 This was the error which I tried to described 我的pip版本:19.2.3 我的python版本:3.7.5 顺便说一句，python sdk安装程序说它； setup(
name
浏览 23提问于2020-05-04得票数 0
1回答
我正在深入研究Flink内部，看看如何在项目中使用它，并向创作者/高级思想家提出了一个问题。为什么Flink使用Yarn作为默认的资源管理器？
库伯奈特斯被考虑了吗？或者这是我们从亚恩开始的事情之一，效果很好.
我遇到过许多项目和文章，这些项目和文章允许Kubernetes和Yarn一起工作，包括允许使用Mesos部署纱线的Myraid项目(但我在Kubernetes.)
我使用了一个非常大的计算集群2000左右的节点，我希望使用Flink的超级酷CEP特性，它依赖于Kafka基础设施(也部署在这个kubernetes环境中)。
我正在寻找理解的原因，使用亚恩作为资源管理下的Flink和如果有
浏览 2提问于2018-03-01得票数 2
回答已采纳
1回答
我正在创建2个flink sql表，1个用于CSV文件系统，另一个用于kafka。其目标是持续监视文件系统文件夹，并将新的csv文件记录推送到kafka主题。但是我在下面写的查询，推送csv文件记录一次，flink作业进入“完成”模式，任何新的文件都不会被处理。请有人告诉我如何创建flink sql连续流与源和csv文件系统和目标为Kafka。
创建源表的Flink SQL
CREATE TABLE son_hsb_source_filesystem_csv_bulk(
file_name STRING,
start_time STRING,
oss_cell_id B
浏览 1提问于2021-03-24得票数 0
回答已采纳
我读取了一个csv文件。我想使用一个长类型的列来翻滚。我使用UDF传输长类型到时间戳类型，但是无法工作错误消息:窗口只能在时间属性列上定义。
我试着调试。TimeIndicatorRelDataType不是时间戳，我不知道怎么转，为什么？
def isTimeIndicatorType(relDataType: RelDataType): Boolean = relDataType match {
case ti: TimeIndicatorRelDataType => true
case _ => false
}
码
public static void mai
浏览 0提问于2019-11-27得票数 2
回答已采纳
我正在开发一个Flink流媒体程序，它可以读取卡夫卡消息，并将消息转储到亚马逊网络服务s3上的ORC文件。我发现没有关于Flink的BucketingSink和ORC文件写入器的集成的文档。并且没有可以在BucketingSink中使用的ORC文件写入器实现。
我被困在这里了，有什么好主意吗？
浏览 0提问于2017-12-06得票数 2
Flink流支持内部连接表达式，如window- join，interval-join.但不支持左联接/完全联接表达式。当然，窗口-cogroup表达式可以实现相同的语义，即使事件立即加入，也必须等待完全窗口大小的时间。我的问题是：
如何从设计角度解释Flink流不支持左连接/完全连接表示？如何通过Flink DataStream API (如果能够立即转发连接事件)来实现它？是否有一种方法可以扩展Flink DataStream API以支持左连接：
.leftJoin()
.where()
.window()
.apply()
浏览 3提问于2021-08-15得票数 0
回答已采纳
我正在尝试找出Apache Flink的关键健康指标是什么。Flink文档提供了大量的指标列表，很难找出重要的指标。
浏览 29提问于2019-02-01得票数 1
回答已采纳
我有一个flink数据流，我正在使用KeyedProcessFunction对其进行一些处理，然后我需要在处理后将数据保存在MongoDB中，但我无法这样做。谁能给我一些关于如何将数据从flink数据流保存到Mongo的输入？
浏览 19提问于2020-12-09得票数 0
回答已采纳
Flink是流媒体的一个明显的选择，然而Spark已经成熟了ML pipelines，它可以在spark中训练模型，将其保存为CrossValidatorModel并部署到Flink中的流媒体数据吗？全部在scala中
浏览 16提问于2017-03-16得票数 0
回答已采纳
我有Mac和m1内核的macbook air。我想从这个docs - operations playground开始使用flink playground (flink 13) docker-compose link进行docker-compose The docker-compose build executed with error
=> ERROR [internal] load metadata for docker.io/library/maven:3.6-jdk-8-slim
浏览 374提问于2021-09-27得票数 0
回答已采纳
1回答
理想情况下，我想运行一些实验来测量我的应用程序的输入使用者的压力(即，如果我的应用程序落后于输入数据主题，并且消息到达的速度快于它们的处理速度)。有人告诉我这很常见，但我不知道该怎么做 我正在阅读1.9版本的Metrics文档，从我得到的内容来看，我必须配置conf/flink-conf.yaml (在独立模式下)，假设jmx报告器是这样的： metrics.reporter.jmx.factory.class: org.apache.flink.metrics.jmx.JMXReporterFactory
metrics.reporter.jmx.port: 8789 那么，我是否应该使用
浏览 51提问于2020-06-15得票数 0
1回答
我知道Flink Default Scheduler策略，但我想要更改某些situations.For示例中的启动策略，例如Io密集型和Cpu密集型，当群集资源在每个node.So中有很大差异时，我想更改默认值是否可以基于群集资源执行我的启动？
浏览 5提问于2018-12-09得票数 0
我试图使用Table在Flink应用程序中利用s3a文件系统。我在EMR6.1(预装有Flink 1.11.0)。
作为应用程序的一部分，我正在s3位置上创建一个表。当我运行这个程序时，我会得到以下错误：
org.apache.flink.core.fs.UnsupportedFileSystemSchemeException
指示s3://或s3a:// URI都不支持任何文件系统。
每一个，如果我在电子病历，我不需要配置任何文件系统作为一个插件。
我反复检查了我的应用程序中没有包含fs-s3-hadoop依赖项。
奇怪的是，当我按照上面的说明将文件系统安装为一个插件时，我可以将它用于s3:
浏览 2提问于2021-09-01得票数 0
回答已采纳
1回答
我尝试使用flink api：
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
// ingest a DataStream from an external source
DataStream<Tuple3<Long, String, Integer>> ds = env.addSource
浏览 4提问于2017-06-08得票数 0
回答已采纳
1回答
根据Flink的文档，我们可以在Kubernetes之上部署一个独立的Flink集群，使用Flink的独立部署，或者使用本地Kubernetes部署在Kubernetes上部署Flink。
文件上说
我们通常建议新的用户使用本地Kubernetes部署在Kubernetes上部署Flink。
是因为本地Kubernetes更容易开始，还是因为独立模式是一种遗产？
在本机Kubernetes模式中，Flink能够根据所需资源动态分配和取消分配TaskManagers。在独立模式下，任务管理器必须手动配置。
在我看来，本地的Kubernetes模式是一个更好的选择。
浏览 7提问于2022-01-05得票数 1
2回答
我在听11/24的Hansel分钟播客谈论月光。
套用一下，Scott评论说，微软对开源的态度正在迅速改变，特别是随着MVC架构的采用。
我承认我对MVC对开发世界意味着什么没有最好的“大局”把握，但我特别想知道它是如何让开源社区受益的？
浏览 6提问于2008-12-30得票数 1
回答已采纳
我有一个ML模型，它需要两个numpy.ndarray -- users和items --并返回numpy.ndarray predictions。在普通Python代码中，我会这样做：
model = load_model()
df = load_data() # the DataFrame includes 4 columns, namely, user_id, movie_id, rating, and timestamp
users = df.user_id.values
items = df.movie_id.values
predictions = model(users,
浏览 6提问于2021-04-14得票数 1
回答已采纳
1回答
我有一个工作正常的嵌入式作业，我想部署更多的同地作业。这些新增的工作将收到来自旧工作的信息，并将其发送到kafka主题。
代码如下
@AutoService(StatefulFunctionModule.class)
public final class CoLocatedModule implements StatefulFunctionModule {
@Override
public void configure(Map<String, String> globalConfiguration, Binder binder) {
FunctionProvide
浏览 2提问于2020-05-03得票数 1
回答已采纳
我们感兴趣的是从连接到常规的Flink流应用程序，最好使用Table。这样做的想法是查阅Statefun在Flink中注册的表，是否有此可能，如何做才是正确的方法？
到目前为止，我的想法是在一些主函数中初始化我的表流，并注册一个有状态函数提供程序来连接到该表：
@AutoService(StatefulFunctionModule.class)
public class Module implements StatefulFunctionModule {
@Override
public void configure(Map<String, String> globalC
浏览 2提问于2020-04-11得票数 2
我有一个Flink 会话 集群在Kubernetes之上，最近我从基于ZK的HA切换到Kubernetes HA。 通读 https://cwiki.apache.org/confluence/display/FLINK/FLIP-144%3A+Native+Kubernetes+HA+for+Flink#FLIP144:NativeKubernetesHAforFlink-LeaderElection 我可以在Flink名称空间上观察到上述文档中描述的每个资源的配置映射： k8s-ha-app1-00000000000000000000000000000000-jobmanager
浏览 38提问于2021-02-12得票数 0
回答已采纳
如何动态更新module.yaml？例如，我想添加新的远程函数，所以我必须更新module.yaml文件。我应该停止/启动flink集群来更新module.yaml吗？什么是真正的用例方法？更新或现有远程函数作业正常。因为滚动更新就足够了。
浏览 24提问于2020-05-18得票数 1
如何将DynamoDB用作AWS数据分析应用程序(Flink)的接收器？
我没有找到示例或现有的DynamoDB Sink实现类。
浏览 26提问于2022-11-22得票数 0
回答已采纳
虽然Flink网站上的所有文档都没有提到“外部状态”，但我们在一篇论文中遇到了这个概念(http://www.vldb.org/pvldb/vol10/p1718-carbone.pdf 4.1节)。我们如何实现这一点？
浏览 26提问于2020-09-03得票数 0
回答已采纳
在部署Flink状态函数时，需要指定函数的端点，即Flink需要命中什么URL才能触发远程函数的执行。
状态：
URL模板名可能包含根据函数的特定类型填充的模板参数。例如，发送给消息类型com.头/greeter的消息将被发送到。
endpoints:
- endpoint:
meta:
kind: http
spec:
functions: com.example/*
urlPathTemplate: https://bar.foo.com/{function.name}
urlPathTemplate支持哪些其他模板值，这些
浏览 4提问于2021-10-18得票数 0
回答已采纳
我刚接触过Flink，目前正在测试一个实用程序的框架，该框架包括丰富来自Kafka的具有许多历史特性的事务(例如，相同来源和相同目标之间过去的事务数)，然后使用机器学习模型对此事务进行评分。
目前，功能都保持在Flink状态，相同的工作是对丰富的事务进行评分。但是我想把功能计算工作和评分工作分开，我不知道如何做到这一点。
可查询状态似乎不适合这一点，因为工作id是必要的，但告诉我，如果我错了！
我考虑过直接查询RocksDB，但也许还有更简单的方法吗？
对Flink来说，将两份工作分开是个坏主意吗？为了避免复杂的作业(并检查它是否对延迟有任何积极的影响)，我们采用这种方法对Kafk
浏览 0提问于2018-02-02得票数 0
回答已采纳
我正在一个包含3个任务管理器的集群上运行一个Flink作业(在3个Kubernetes吊舱之上)。作业的默认并行度为9，其中一个操作符设置为并行18。作业的任务槽数设置为18 (最大并行值)。
我观察到以下行为：
设置为并行性18的运算符在所有任务时隙之间平均分布。
所有其他运算符(设置为默认- 9)都不是平均分布的。例如：
TM1:运行两个子任务
TM2:运行5个子任务
TM3:运行两个子任务
有谁能解释一下-
是什么导致了这种不均衡的分布？
我能控制操作者的任务吗？我该怎么做呢？
(使用Flink v1.6.3运行)
浏览 0提问于2019-03-26得票数 3
回答已采纳
我认为斯派克比弗林克优越。下面是我的研究。
我看到星火的大部分功能都包含在Flink中，除了星火的“公平调度”。我试着搜索并查看Flink文档，但是没有运气。
另外，如果你看到吉顿，Apache的流行度几乎是Flink的两倍(明星和叉子的数量)。所以我很想知道为什么弗林克没有星火那么受欢迎。
我还看到写入Flink的连接器数量比Spark的连接器数量少/维护少。(例如: mongodb)这是否意味着Flink尚未成熟或获得市场吸引力？
以上问题的答案将帮助我们决定合适的技术。
编辑-1
在阅读了一些答案之后，我会给出更多的信息。
这将主要用于批处理。实时流可能占用例的1
浏览 4提问于2022-04-21得票数 -2
我想设置Flink，这样它就可以将数据流从Apache转换并重定向到MongoDB。出于测试目的，我构建在flink流-Connectors.kafka示例()之上。
卡夫卡流是正确红色的链接，我可以映射他们等，但问题发生时，我想保存每一个接收和转换的信息到MongoDB。我发现的关于MongoDB集成的唯一例子是来自github的flink-mongodb测试。不幸的是，它使用的是静态数据源(数据库)，而不是数据流。
我认为应该有一些DataStream.addSink实现的MongoDB，但显然没有。
实现这一目标的最佳途径是什么？我是否需要编写自定义接收器函数，还是可能遗漏了什么？也许应
浏览 3提问于2016-02-02得票数 11
回答已采纳
1回答
站点将几个项目列为上游项目。有90多个项目，看起来不像JBOSS (Camel、Cxf、tomcat等)。
这个词是什么意思？
浏览 3提问于2015-01-29得票数 0
回答已采纳
1回答
我是否使用了正确的框架？
、、、
我刚开始使用scala / flink / spark，我想问几个问题。现在正在使用带有flink的scala。
数据流的一般思想如下：
csv文件-> flink ->弹性-> flink(工艺数据) -> MongoDB -> Tableau
有大量的日志文件是分号分隔的。我想将这些文件写入elasticsearch作为我的数据库。(这已经奏效了)
现在需要进行各种分析(f.e。一致性报告/生产力报告)。对于这些报告，需要不同类型的列。
其思想是使用flink从elasticsearch导入基本数据，编辑数据并将其保存到mongodb中，这样就可以使用tabl
浏览 1提问于2017-08-08得票数 2
回答已采纳
1回答
我是flink的新手，我们的用例处理无状态计算。读取事件、处理事件并保存到数据库中。但是Flink文档从来没有提到过无状态处理。任何示例存储库，用于查找无状态示例或文档。
最后，对于这个用例，哪个Flink模型有效？流应用程序或事件驱动应用程序。
浏览 21提问于2020-12-09得票数 2
我正在Kubernetes上编写一个Flink应用程序，它可以读写Kafka，所以我需要一个Kafka集群来进行测试。我只是想找到在Kubernetes中运行Kafka集群的最简单的方法，它可能只是一个单节点集群，性能不是问题。
浏览 1提问于2019-10-24得票数 0
我正在使用Flink，并希望将结果值插入Redis。当我搜索Redis时，我发现了Apache中包含的redis-connector。因此，我可以使用Apache中的reids-connector将结果值插入到redis中。
但是，我认为我也能够使用Jedis连接到Redis。我进行了实验，表明我能够连接redis，并使用jedis找到插入到redis中的值，如下面的代码所示。
DataStream<String> messageStream = env.addSource(new FlinkKafkaConsumer<>(flinkParams.getRequir
浏览 7提问于2022-11-03得票数 0
1回答
谁能告诉我使用SNMP监控Docker容器是不是一个好主意？我的意思是，我正在考虑在每个容器上安装SNMP代理，并通过Flink/Kafka流收集数据，但我不知道这样做是否可以，就像在每个容器上安装SNMP代理一样。
谢谢!
浏览 1提问于2018-07-13得票数 1
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM
腾讯云开发者
