如何在运行时从flink中的数据流映射或flatMap执行sql查询
在运行时从Flink中的数据流映射或flatMap执行SQL查询，可以通过以下步骤实现：
- 理解Flink的基本概念：
- Flink是一个流式处理框架，用于处理无界和有界数据流。
- Flink提供了SQL API，可以在流处理中执行SQL查询。
- 创建Flink的执行环境：
- 首先，需要创建一个ExecutionEnvironment或StreamExecutionEnvironment对象，用于定义Flink作业的执行环境。
- 定义数据源：
- 使用Flink提供的DataStream API，从外部数据源（如Kafka、文件、Socket等）读取数据流，并将其转换为DataStream对象。
- 执行SQL查询：
- 使用Flink的Table API或SQL API，将DataStream注册为一个表，并执行SQL查询。
- 可以使用Table API的map()或flatMap()函数将数据流映射到表中的列。
- 可以使用SQL API的SELECT语句执行SQL查询。
- 处理查询结果：
- 根据查询的需求，可以使用Flink提供的各种操作符（如filter、groupBy、join等）对查询结果进行进一步处理。
- 可以将处理后的结果输出到外部系统（如数据库、文件、消息队列等）或继续在Flink中进行后续计算。
以下是一些腾讯云相关产品和产品介绍链接，可用于支持上述步骤中的各项操作：
- 数据源：
- 腾讯云消息队列 CMQ：提供可靠的消息传递服务，可作为数据流的来源。
- 产品介绍链接：https://cloud.tencent.com/product/cmq
- 执行环境：
- 腾讯云弹性MapReduce（EMR）：提供大数据处理的执行环境，支持Flink。
- 产品介绍链接：https://cloud.tencent.com/product/emr
- 数据存储：
- 腾讯云对象存储 COS：提供高可靠、低成本的对象存储服务，可用于存储Flink的输入和输出数据。
- 产品介绍链接：https://cloud.tencent.com/product/cos
- 数据库：
- 腾讯云云数据库 MySQL：提供高性能、可扩展的关系型数据库服务，可用于存储和查询Flink的结果数据。
- 产品介绍链接：https://cloud.tencent.com/product/cdb
请注意，以上链接仅供参考，具体选择适合的产品和服务应根据实际需求进行评估和决策。
相关·内容
我正在开发一个SSIS包，它有一个数据流任务，它执行从SQL查询获取数据并将其插入SQL表的工作，下面的属性来自用户定义的变量，这些变量反过来从Server表中获取值，数据流任务在for每个循环中执行表中的所有记录。
源连接
源查询
目的连接
目标表
元数据不会刷新新连接/查询/目标表，有什么方法可以在运行时自动处理吗？
浏览 1提问于2016-03-30得票数 0
回答已采纳
我想知道是否有任何方法可以在运行时在Apache Flink中的一个操作符内的并行执行之间传输信息？我只需要发一点信息。Flink中的广播变量不起作用，因为它不能在运行时广播。
浏览 6提问于2018-02-24得票数 0
2回答
如何在Flink转换中更新配置？
、、、、
给定一个Flink流作业，该作业将map()操作应用于流。
此map()操作从某些属性读取其配置，并相应地映射数据。例如，配置指定读取属性"input"，并使用不同的属性名"output“将其写入流。这个已经很好了。
现在，配置发生了变化，例如，转换将为输出使用不同的属性名。
因此，我正在寻找一种方式，让所有的Flink任务重新阅读一个新的配置在运行时。
有没有可能
将KafkaSourcewait挂起，直到管道耗尽(flush)trigger集群中的所有任务，以重新读取配置文件(coordinated)resume (coordinated)resume KafkaS
浏览 13提问于2022-10-11得票数 0
在处理每个数据流的多个数据流和多个操作符时，我试图理解Flink的运行时方面。
用例:单个flink作业中的n个数据流(每个数据流代表一个设备--具有不同的时间延迟)，每个数据流被分成两个流，其中一个流被分成几个CEP操作符，另一个流为一个进程函数。
问题：
在运行时，引擎会在每个数据流中创建一个线程吗？或者每个操作符一个线程？
在作业开始时，可以在运行时动态创建数据流吗？(也就是说，当作业开始时从文件中读取N，并且需要创建相应的N个流)
当创建大量流(N ~ 10000)而不是单个流中的N个分区时，是否存在特定的性能影响？
浏览 2提问于2017-11-09得票数 1
我正在学习Flink，有一件事让我感到困惑，那就是使用了一个名为Collector的对象。例如在平面映射函数中。这个收集器及其方法收集了什么？例如，为什么map函数不需要通过显式使用它来传递结果呢？
下面是在平面地图函数中使用收集器的一些示例：
此外，如果我搜索收集器在Flink Architecture中的放置位置，我也找不到任何具有该映射的图
浏览 58提问于2021-02-20得票数 1
我计划创建一个具有flink流的规则引擎。
在执行过程中有一些要求：
所有要根据规则执行的事件都必须从卡夫卡中读取。
所有规则都必须在有限的时间内执行。
问题是可以在运行时添加规则，所以我不能简单地创建大量作业来处理所有传入消息，因为我冒着超过执行规则的最大时间的风险。
我有一个保证，一个单一的规则可以在时限内执行。
所以，我不知道这是不是一个好的做法，创造一个职位的规则，并增加更多的就业机会时，新的规则即将到来？(这可能是上百条规则)。
我有一种直觉，认为这不是解决问题的方法，但没有什么真正合理的解释原因。
第二种方法是维护队列(例如，在动物园管理员中)，以跟踪针对哪个事件执
浏览 0提问于2016-10-04得票数 3
回答已采纳
以下用例的最佳实践建议是什么？我们需要将流与一组“规则”相匹配，这些“规则”本质上是一个Flink DataSet概念。对此“规则集”的更新是可能的，但并不频繁。每个流事件必须对照“规则集”中的所有记录进行检查，而且每次匹配都会将一个或多个事件生成到接收器数据流中。规则集中的记录数在6位范围内。
目前，我们只是将规则加载到本地规则列表中，并在传入的flatMap上使用DataStream。在flatMap中，我们只是迭代一个列表，将每个事件与每个规则进行比较。
为了加快迭代速度，我们还可以将列表分成几个批，实质上是创建一个列表列表，并创建一个单独的线程来迭代每个子列表(使用Java或Scala
浏览 1提问于2018-06-04得票数 0
回答已采纳
我试图转换数据流，而不使用Flink提供的任何窗口。我的代码如下所示：
val stream1 = executionEnvironment.getStream
val stream2 = stream1.flatMap(someFunction)
stream2.addSink(s3_Sink)
executionEnvironment.execute()
但是，在提交和运行我的工作时，我无法在S3上获得任何输出。web显示接收到的0字节、接收到的记录、发送的0字节、发送的记录。
另一个正在运行的Flink作业已经使用了相同的数据源，因此数据源很好。任何地方都没有错误，但仍然没有输出。这个
浏览 0提问于2019-04-03得票数 2
回答已采纳
这是我的Flink SQL
SELECT t.reportCode FROM query_record_info as t LEFT JOIN credit_report_head as c ON t.reportCode = c.reportCode
当我运行它时，我搞错了
Exception in thread "main" org.apache.flink.api.common.InvalidProgramException: Selected sort key is not a sortable type
at org.apache.flink.api.java
浏览 8提问于2020-08-11得票数 0
回答已采纳
我的前提服务器上有50张桌子。我想在数据砖中将这50个表从前提表迁移到增量表。但是每个表都定义了特定的模式，但我需要设计一个adf管道，以便将这50个表从现场移动到delta表。
如何根据表名在运行时将架构附加到数据帧？
浏览 2提问于2022-01-28得票数 0
我一直试图在Flink中使用SOURCE语句创建一个文件。从医生那里：
源从文件中读取SQL SELECT查询并在Flink集群上执行该查询。
它将用于Flink版本1.13.1，Scala2.11。据我所知，关于这个Flink发行版的其他一切都很好。
在集装箱内：
/opt/flink# echo "select 'hello';" > test.sql
/opt/flink# sql-client.sh
然后在sql客户机中：
Flink SQL> select 'hello';
-- works as intended
浏览 8提问于2021-06-11得票数 0
回答已采纳
在使用ADF插入数据之前，我尝试运行一个预SQl脚本。我的预SQL脚本包含数据流参数。请注意下面的参数值。
parameter1 = excel工作表中的一列(假设值为'XY')
PFB查询
“从ABC删除col_name = {$parameter1}”-输出:：XY
“从ABC删除col_name = '{$parameter1}'”
当我在查询下面运行时
“从ABC删除col_name ='{$parameter1}'”--错误:文字表达式中不允许列操作数
如果我在下面尝试
concat(从ABC中删除col_name ='，toS
浏览 3提问于2021-06-08得票数 1
回答已采纳
是否可以在外部查询现有的StreamTable，而不上传.jar获取执行环境和检索表环境？由于Apache表1.4的动态(连续)表特性，我一直在等待它的发布。我期待其他的东西，我认为在运行时修改表并修改它的参数是可能的。为了执行一些活动查询，而不是在数据流之上定义(连续的或仅附加的)数据库视图。我知道我可以将我的表导出到一些数据库中，并使用SQL动态查询这个数据库，但这是一种尴尬的行为。Flink的美妙之处在于，一切都是实时的，所有的东西都是流，那么是否可以从外部程序实时查询Flink表呢？
浏览 5提问于2017-10-25得票数 2
回答已采纳
我完全是阿帕奇·弗林克的菜鸟。只是想弄脏我的手。我有下面的场景。
事件数据报
事件数据报
数据交换规则
结合这两种基于ruleID的数据流
现在我有了一个tuple3的数据仓库，它看起来像<ruleId, Rule, Event>。这些规则是我希望在事件上运行的SQL查询。
我正在学习动态表和Flink SQL的概念。我不知道如何进一步处理。有人能帮帮我吗？
浏览 0提问于2018-02-09得票数 0
我已经编写了flink作业来使用数据集和数据流api。我已经从相同的main()方法启动了这两个程序，但现在当我将作业提交到flink Dashboard时。只有dataset api程序正在运行，仪表板显示作业已完成，流未被触发，flink仪表板也未显示有关数据流执行的任何内容。但是，当我从eclipse运行时，数据集和数据流api程序都在运行。是否有其他方法可以将作业提交给flink作业管理器以同时完成这两项工作？
浏览 16提问于2017-07-12得票数 1
我正试图在官方的帮助下将Flink流处理器部署到Kubernetes集群。Flink应用程序还使用Minio作为其状态后端。在我试图以下列方式提供Hashicorp Vault的凭据之前，一切都很好：
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: flink-app
namespace: default
spec:
serviceAccount: sa-example
podTemplate:
apiVersion: v1
kind: Pod
meta
浏览 20提问于2022-09-02得票数 3
2回答
Flink的嵌套输出
、、、
我正在使用Flink SQL处理Kafka流，其中每条消息都是从kafka中提取的，使用flink sql处理，然后推回到Kafka中。我想要一个嵌套的输出，其中输入是扁平的，输出是嵌套的。例如，我的输入是 {'StudentName':'ABC','StudentAge':33} 并希望输出为 {'Student':{'Name':'ABC','Age':33}} 我试着搜索here和一些类似的链接，但没有找到。使用Apache Flink SQL API可以做到这一点吗？如有必
浏览 56提问于2019-01-04得票数 0
我正在尝试创建一个SSIS包，它将把csv数据导出到excel文件。这个包将从C#调用，将csv文件作为包的输入。我对SSIS非常陌生，并且能够用相同的头生成预期的结果。
我采用了以下方法-
脚本任务-从csv头创建脚本来创建临时表、大容量插入、excel表脚本。
执行SQL任务-在数据库中创建临时表
执行SQL批量插入csv数据到表中
执行SQL任务-创建Excel文件
数据流任务- OleDB源到Excel目标
执行SQL任务-删除创建的临时表
我面临的挑战是我的csv可能有不同的标题(文本和标题的数目都可能不同)。我想要一个包裹来达到这个目的。
在头不同的
浏览 0提问于2013-02-18得票数 0
在单个INSERT INTO ... SELECT ...作业中(在Flink 1.13.6上)是否可能有多个PyFlink语句？
我创建了许多输出表，并试图在单个作业中写入这些表，其中Python & SQL示例(假设有一个名为“input”的输入表)：
sql1 = "INSERT INTO out1 (col1, col2) SELECT col1, col2 FROM input"
sql2 = "INSERT INTO out2 (col3, col4) SELECT col3, col4 FROM input"
env.execute_sq
浏览 8提问于2022-04-29得票数 0
回答已采纳
我试着用delta表作为我的水槽，我想把真空设置为0。它抛出了一个错误，我应该首先设置配置'spark.databricks.delta.retentionDurationCheck.enabled':'false‘，但是我看不到在哪里设置这个属性，因为用于映射数据流的计算集群是由Azure在运行时创建和管理的。另外，我还想根据需求配置另一种星星之火，即‘scapk.sql.parquet.pression.codec’。有可能吗？
浏览 2提问于2021-12-02得票数 0
我正在从flink表api编写一个查询来检索一条记录。然后检查是否找到一条记录，如果找到，则获取该记录的每个列值的字符串值。
即
users:
|id | name | phone |
|---|------|-------|
| 01| sam | 23354 |
| 02| jake | 23352 |
| 03| kim | 23351 |
问题是flink只从查询返回Table，所以我不能1:检查是否找到一条记录，2:获取找到的记录值的各个值
sudo代码：
foundRecord = find record by phone
if foundRecord {
cre
浏览 4提问于2020-05-30得票数 0
回答已采纳
2回答
我正在尝试从运行在clojure中的Apache中获得一个小示例，但是现在我被困住了，因为clojure中的类型暗示和flink中的一些奇怪的怪癖。
这是我的代码：
(ns pipeline.core
(:import
(org.apache.flink.api.java ExecutionEnvironment)
(org.apache.flink.api.common.functions FlatMapFunction)
(org.apache.flink.api.java.tuple Tuple2)
(org.apache.flink.util Collector)
(jav
浏览 2提问于2015-08-20得票数 5
回答已采纳
2回答
我有一个需求，需要在ADF管道中映射数据流与SQL存储过程之间进行选择，以实现一些业务场景。现在的数据量并不太大，但在稍后阶段可能会变大。业务逻辑有时很复杂，我必须连接多个表、编写子查询、使用windows函数、嵌套case语句等等。
我的所有业务需求都可以很容易地通过SP实现，但是考虑到数据流在下面运行，并且可以根据需要进行扩展，所以有一点倾向于映射数据流。ADF映射数据流在ADF管道中使用时是否优于SQL存储过程？我对映射数据流的一些关注如下。
使用数据流实现复杂逻辑所需的时间比存储过程要长得多，映射数据流的执行时间要高得多，考虑到旋转星星团所需的时间。
现在，如果我决定在管道中使用SQL
浏览 7提问于2020-09-15得票数 3
2回答
如何在SSIS中调用存储过程？
、、、、
我正在尝试创建一个SSIS包，该包从一个表中查询数据，并调用另一个数据库中的存储过程，每行调用一个存储过程。
在我的旧DTS包中，我是这样做的：
EXEC myStoredProcedure ?, ?, ?
...and，然后我映射了参数。但是，在SSIS中，我不知道如何实现这一点。
我有一个数据流任务，它首先对数据运行查询。它将数据传递到OLE DB目标。我将数据访问模式设置为"SQL command"，但是当我尝试放入上面的SQL时，当它解析SQL时，我得到了"Invalid Parameter Count“。我无法进入映射屏幕。有什么想法吗？
浏览 0提问于2009-04-13得票数 23
回答已采纳
1回答
我在服务器上有一个存储过程，它在我的数据库中生成一个表。然后在ssis中，我从那个表中查询一些列，然后追加一些填充了静态值的虚拟列。当我查询数据库时，我将查询保存为变量(sql命令从变量)，在该查询中，我使用来自X的select a、b、c，其中@User：：variable 1=‘和@User：：variable 2=’用于所有4。我的问题是:我需要能够将这些变量的值(variable1改为4)更改为48种不同的场景(或可能更多)，因此手动替换它们将很痛苦，因为它将导致超过130个组合体。如果有一种方法可以在运行时将excel文件的值传递给包？例：
column1 column
浏览 3提问于2015-06-24得票数 0
回答已采纳
我正在尝试将数据从存储帐户获取到server。在这方面，我使用ADF数据流来使用自托管运行时的接收器目标系统来连接前提SQL数据库。如果我正在验证我的数据流，获取以下警报消息。请帮帮我。
我的项目架构是(CRM ) azure突触链接->加载数据(存储gen2) -> ADF (数据流)->前提SQL验证警报消息如下：
数据流中不支持具有自托管集成运行时的链接服务。请使用托管vnet的Azure IR
如果没有自托管运行时，我们无法连接外部网络，请建议，我们如何解决它。
浏览 8提问于2022-09-10得票数 0
我正在尝试(未成功)在Apache Flink中运行一个简单的hello world类型的程序。代码从Apache Kafka获取一条消息，并添加一个“。并将新字符串打印到stdout。代码正确地从Kafka获取消息，但是map函数添加了".“失败。我在REPL提示符下尝试了这个函数，scala代码在那里可以正常工作。Scala代码：
scala> input = "hello"
input: String = hello
scala> val output = input.flatMap(value => value + ".")
o
浏览 1提问于2016-11-22得票数 0
1回答
数据流问题
、、
尝试在SSIS包中运行Sql命令时出错。
任务: DataFlow任务
连接: ADO.NET
数据访问模式: Sql命令
select:从表中选择field1 =？field2 =？
错误：“一个或多个所需参数没有给定值”
更多信息：
在包中执行Sql任务：
(常规选项卡)-连接: ADO.NET - SQL语句：exec storedprocedureX ?,?
(参数映射选项卡)
User::field1 , Input , String , 0 , -1
User::field2, Input, String, 1, -1
包中的变量集
field1
浏览 0提问于2019-05-17得票数 2
回答已采纳
我有两个流，一个是主流，比如说在欺诈检测的例子中，我有事务流，然后我有第二个流，这是信任，在我们的例子中它是规则。因此，我将主流连接到配置流，以便进行处理。但是，当flink第一次启动时，当我们添加作业时，它开始从事务中消费，并并行地信任流；当需要进程事务时，它有时会看到没有配置，我们不得不将事务发送到死信队列。然而，我想要实现的是，如果有专利配置，稍后我可以得到，我想先得到配置，然后得到事务，以便处理它，而不是把它发送到死信队列。我有同样的交易和信任的钥匙。
长话短说，是否有办法告诉flink，当第一次作业开始尝试消耗一个流，直到没有新的值，然后开始处理主流流？我怎么能把它们按顺序排列呢？
浏览 13提问于2022-04-14得票数 1
1回答
我正在尝试使用此代码在运行时停止SSIS包
RunningPackages pkgs = app.GetRunningPackages("myserver");
foreach (RunningPackage p in pkgs)
{
Console.WriteLine("PackageName: " + p.PackageName);
p.Stop();
}
运行时无错误
但是，当我调试到在停止每个包和在SQL SERVER Managment Studio中查看正在运行的包之间暂停时，我仍然看到所有的包都在运行。
我甚至尝试在运行的包上单击
浏览 5提问于2010-11-25得票数 4
回答已采纳
我的问题可能会引起一些混乱，所以请先看描述。找出我的问题可能会有帮助。我将在问题的末尾添加我的代码(关于我的代码结构/实现的任何建议也是受欢迎的)。，谢谢您的帮助！
我的问题：
如何在Flink批处理中定义多个接收器，而不让它从一个源重复获取数据？
createCollectionEnvironment()和getExecutionEnvironment()有什么区别？我应该在当地环境中使用哪一种？
env.execute()的用途是什么？我的代码将输出没有这个句子的结果。如果我加上这句话，它会弹出一个例外：
-
Exception in thread "main&
浏览 1提问于2018-06-28得票数 2
1回答
导出日期计算
、、
我目前正在使用SSIS将数据输入Server数据库。计划是每周这样做，但发生的日期可能有所不同，这取决于什么时候数据将被推送。
我使用SSIS从Excel工作表中获取数据，并将每一行输入数据库(大约每周150行)。唯一的公分母是所有行之间的日期。我想在推送日期的当天向每一行添加一个日期。由于推送日期可能不同，所以不能使用从为该行输入的前一个日期起的一个星期的当前日期。
但是因为大约有150行，所以我不知道如何实现这一点。如果我能够在SQL Server中设置它，那么如果每次输入一组新行时，它都会从前一组行中添加7天，那就太好了。但我也很乐意在SSIS中做这件事。
有没有人知道如何做到这一点？或
浏览 2提问于2014-12-29得票数 0
回答已采纳
我已经部署了一个SSIS包，它有一个从oracle中提取数据的查询。查询中有一些变量，我想看看在运行时输入了哪些变量。有没有办法可以通过sql事件探查器，在sql作业运行时捕获sql server发送给Oracle的查询？
浏览 1提问于2014-05-02得票数 1
1回答
我有一个在Sql Server Management Studio中返回有效结果的mdx查询，并希望自动执行此查询并将结果发送到电子邮件中。
SSIS似乎很适合这一点。我已经能够在“执行SQL任务”中运行mdx，并使用结果填充对象变量，但我不确定如何将该对象解析为可在发送邮件任务中使用的字符串或xml数据。我在其他地方看到的关于这类事情的帖子最终都会使用“脚本任务”，这需要我将我的C#思想转换为VB.NET，并且似乎超出了SSIS的精神和效用。
如果有另一种完全可以工作的方法，我不会选择SSIs，但我希望继续使用MS SQL Server和.NET工具集。
浏览 0提问于2009-06-23得票数 0
我正在尝试从kafka读取json数据并在Scala.I中处理它。我对flink和kafka流很陌生，所以请尝试通过给出解决方案代码来回答。我希望能够将它转换为包含所有键和值对的Map。
map1.get("FC196")应该让我处于休眠状态，map1是包含键值对的映射
我面临的挑战是将代码中的第一个变量DataStreamObjectNode转换为键值对的映射。我正在使用JSonDeserializerSchema.If，我使用简单的Schema，我得到了DataStreamString。我愿意接受其他建议。
卡夫卡的输入格式：
{"FC196":"D
浏览 6提问于2017-04-25得票数 0
回答已采纳
我们使用Flink Table API在Flink应用程序中使用Kafka主题。
当我们第一次提交应用程序时，我们首先从自定义注册表中读取最新的模式。然后使用Avro模式创建Kafka数据流和表。我的数据序列化程序的实现的工作方式类似于Confluent模式注册表，它先检查模式ID，然后使用注册表。因此，我们可以在运行时应用正确的模式。
但是，我不知道如何在不重新部署作业的情况下更新表模式和重新执行SQL。有没有办法让后台线程检查模式更改，如果有更改，则暂停当前执行，更新表模式并执行SQL。
这对于连续交付对应用程序的模式更改特别有用。我们已经进行了兼容性检查。
浏览 2提问于2020-11-14得票数 0
我正在启动一个新的Flink应用程序，以允许我的公司执行大量的报告。我们有一个现有的遗留系统，所需的大部分数据都保存在Server数据库中。在开始使用新部署的Kafka流的更多数据之前，我们首先需要使用这些数据库中的数据。
我花了很多时间阅读Flink书籍和网页，但我有一些简单的问题和假设，我希望你能帮助我，这样我才能进步。
首先，我想使用DataStream API，这样我们既可以使用历史数据，也可以使用实时数据。我不认为我想使用DataSet API，但我也不认为使用/Table有什么意义，因为我更喜欢用Java编写函数。我需要维护自己的状态，看来DataStream键控函数是可行的。
现
浏览 1提问于2021-08-10得票数 2
回答已采纳
我需要在K8上部署Flink应用程序的帮助 我们有3个源，将以SQL查询的形式发送触发条件。总查询数约为3-6k，有效地增加了flink实例的负载。我试着执行，但它非常慢，并且需要很多时间才能开始。 由于大量的查询，我们决定为每个源创建多个flink应用实例。因此，一个flink实例实际上只能执行~1-2K个查询。 示例: sql查询源为A、B、C Flink实例： App A -->将只负责处理源A查询 App B -->将只负责处理源B查询 App C -->将只负责处理源代码C查询 我想在Kubernetes上部署这些实例 问题： a)可以使用迷你集群(内置)部署独立
浏览 31提问于2020-08-24得票数 1
3回答
我想知道我是否可以改变已经提交给Flink的作业的操作人员。假设我有一个单词计数程序，并且上面有一个过滤器，只对大于3个字符的单词进行计数。我希望在运行时更改此筛选器的参数。我的第一个猜测是Flink (以及其他数据流引擎--火花、Storm、Apache )无法做到这一点，因为这项工作已经在env.execute()上提交了。有人知道做这件事的方法吗？
我想这个问题()是与我想要的相关的，但是解决方案仍然不是我想要的动态的。
谢谢
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironme
浏览 0提问于2018-12-20得票数 1
回答已采纳
我正在运行flink run-application目标Kubernetes，使用以下选项：
-Dmetrics.reporter.prom.class=org.apache.flink.metrics.prometheus.PrometheusReporter
-Dmetrics.reporter.prom.port=9249
我指定了一个容器镜像，它将普罗米修斯插件复制到/opt/flink/plugins中。从作业管理器容器中，我可以在端口9249上下载Prometheus指标。但是，flink pod上的kubectl describe未显示Prometheus端口已暴露。kubect
浏览 17提问于2021-06-22得票数 0
Pattern< Tuple3< String, String, String >, ? > pattern = Pattern.<Tuple3< String, String, String > > begin( "start" )
.next( "3" ).where( new FilterFunction< Tuple3< String, String, String > >() {
@Override
浏览 9提问于2017-01-10得票数 0
我是Apache Flink的新手。我使用Pojo Sink将数据加载到Cassandra。现在，我在@ table 注释的帮助下指定表名和键空间名。
现在，我希望在运行时动态地传递表名和键空间名，以便将数据加载到用户指定的表中。有办法做到这一点吗？
浏览 3提问于2016-12-16得票数 4
1回答
Azure Data数据流的成本是多少？账单是否取决于我们创建的数据流/集成运行时？此外，账单是否取决于我们从创建ADF数据流时开始运行管道还是开始计费？如果是这样的话，在Azure中是否有禁用/暂停ADF数据流帐户的选项？
浏览 4提问于2020-04-30得票数 2
回答已采纳
是否可以在运行时更改flink框架的日志级别？什么时候flink集群已经在运行？我参考了下面的文件，没有提到它。 https://ci.apache.org/projects/flink/flink-docs-stable/monitoring/logging.html
浏览 267提问于2020-11-24得票数 0
我在eclipse中运行flink，Maven已经获取了必要的jars。我的机器有一个8核的处理器，我必须编写的流媒体应用程序从它的输入中读取行，并计算一些统计数据。
当我在我的机器上运行这个程序时，我希望flink使用CPU的所有核心作为线程代码。但是，当我查看内核时，我发现只有一个内核在使用。我尝试了很多东西，并在下面的代码中留下了我最后一次尝试的内容，即设置环境的并行性。我还尝试为流单独设置它，依此类推。
public class SemSeMi {
public static void main(String[] args) throws Exception {
浏览 0提问于2016-01-15得票数 1
1回答
我需要将所有SQL查询存储在一个文件夹下，并在SSIS包中引用，以便更好地组织我正在使用的SQL文件列表，这样我们以后就可以轻松地更改SQL文件，而不必重新构建包。这将包括我用于“执行SQL任务”的所有查询，以及“数据流组件”下的"OLE DB数据源“中的查询。
在数据流任务下，不是将源数据库的SQL查询放到查询窗口中:我在OLE DB数据源的数据访问模式下看到了四个选项- 1.表或视图2.视图表名称变量3. SQL命令4.来自变量的SQL命令
我知道我可以使用一个变量，将查询存储在变量中，并在“执行SQL任务”中引用它，但我正在寻找一种方法来存储SQL文件中的所有查询，并将其存储在数
浏览 0提问于2016-05-30得票数 0
我有一个SSIS包，它从SQL表中获得一个新的GUID列表。然后，我将GUID分解到一个字符串变量中，以便用逗号将它们隔开。它们在变量中的显示方式示例如下：
'5f661168-aed2-4659-86ba-fd864ca341bc','f5ba6d28-7283-4bed-9f11-e8f6bef225c5'
问题出在数据流任务中。我在SQL查询中使用该变量作为参数来获取源数据，但无法获得结果。当WHERE子句看起来像：
WHERE [GUID] IN (?)
我得到一个无效的字符错误，所以我发现隐式转换不能像我想象的那样与GUID一起工作。如果这是一个单独
浏览 3提问于2012-07-19得票数 3
回答已采纳
我只想我的flink应用程序和可配置的一样多。此外，我希望在运行时更改进程函数的行为，而不是停止集群并重新部署jar文件。
有那份文件吗？或者是否可以将进程函数代码注入正在运行的jar中。例如，从web中，我将获得流程函数输入(作为java代码)，然后在提交表单后，我将更新流程函数行为。
浏览 2提问于2022-08-15得票数 2
我正在尝试将Oracle数据库表中的记录合并到本地SQL表中。
我有一个包的变量，它是一个对象，名为OWell。
我有一个数据流任务，它以SQL语句的形式获取OWell数据(select order by order by well_id，select well_id，order from order by Well_ID)，然后还有一个转换任务，将well_id从长度为15的DT_STR转换为DT_WSTR；将well_name从长度为15的DT_STR转换为长度为50的DT_WSTR。然后将其存储在记录集OWell中。
进行转换的原因是我要向其中添加记录的表有一个标识字段: SSIS将we
浏览 0提问于2010-05-19得票数 3
回答已采纳
亲爱的所有人，
我在Flink客户机上查询hive表，返回的结果是all NULL。这个蜂箱表是一个或ORC表，.I不知道为什么
Flink SQL>从table1选择*，其中dt=‘1609739880002’；
Flink SQL>从table1中选择计数(*)，其中dt='1609739880002'；
浏览 2提问于2021-01-04得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储活动推荐
腾讯云开发者
