如何在Flink行中通过FieldName获取值？
在Flink中，可以通过FieldName获取值的方法是使用Flink的Table API或DataStream API中的相应函数和操作符。
- 使用Table API：
- 首先，将输入数据源转换为一个Table对象。
- 然后，使用Table对象的select()方法选择需要的字段。
- 最后，使用字段的名称作为参数调用getField()方法来获取字段的值。
- 示例代码如下：
- 示例代码如下：
- 使用DataStream API：
- 首先，使用map()或flatMap()等转换操作将输入数据流转换为一个新的数据流。
- 然后，使用getField()方法并传入字段的索引或名称来获取字段的值。
- 示例代码如下：
- 示例代码如下：
请注意，上述示例代码中的"fieldName"应替换为实际的字段名称。另外，Flink提供了丰富的函数和操作符，可以根据具体需求进行更复杂的数据处理和转换操作。
关于Flink的更多信息和相关产品介绍，您可以参考腾讯云的官方文档：
相关·内容
我有一个def process(row: org.apache.flink.types.Row, fieldName: String) : Unit = ???方法
我要做的处理需要提取一个名为fieldName的字段。我可以看到Row api允许您按顺序提取，但不允许按名称提取。
我猜用户必须向我传递模式信息，所以这个方法看起来像def process(row: org.apache.flink.types.Row, rowSchema: ???, fieldName: String) : Unit = ???。
我已经在org.apache.flink.table.types.logica
浏览 30提问于2020-05-06得票数 1
回答已采纳
2回答
我已经声明了布尔标志如下。
[Flags]
public enum StudentStatus
{
True = 1,
False = 2
}
我将通过下面一行的DataValues集合获取值，并将其赋值到下面的属性中。
var student= new Student();
student.Status= StudentInfo.Data.DataValues
.Where(m => m.FieldName.Equals("Status"))
.S
浏览 5提问于2014-08-09得票数 1
回答已采纳
是否可以在运行时更改flink框架的日志级别？什么时候flink集群已经在运行？我参考了下面的文件，没有提到它。 https://ci.apache.org/projects/flink/flink-docs-stable/monitoring/logging.html
浏览 267提问于2020-11-24得票数 0
我正在使用flink从Azure数据湖中读取数据。但flink无法找到Azure数据湖文件系统。我已经实现了spark来读取Azure Data Lake文件系统文件。
因此，我尝试从flink中触发火花，从Azure数据湖中读取数据，并执行一些逻辑操作，然后将列表返回给flink。所以flink可以使用这些数据？
是否可以从flink触发spark作业？或者如何配置flink来理解Azure数据湖文件系统。有人能在这方面给我指点一下吗？
浏览 2提问于2017-07-11得票数 0
DataGrid列是在后面的代码中动态创建的。为了动态绑定列，我有一个Dictionary<string, obj>属性，其中键作为列标题，值作为值。
列被绑定到字典，如下所示：
var item = new DataGridTextColumn();
item.Width = new DataGridLength(1, DataGridLengthUnitType.Auto);
item.Header = name;
item.Binding = new Binding($"di[{name}].obj") { Mode = BindingMode.TwoWay
浏览 12提问于2017-07-16得票数 1
1回答
标题说明了一切，但我将更详细地解释。事情是这样的。
我目前正在开发Word VSTO插件中的额外代码，该插件在模板中使用合并字段。模板中的合并字段使用外部文件中的数据填充。我需要的是从合并字段中读取值，但我完全不知道如何实现这一点。我已经搜索了几天了，但我读的文章都没有对我起作用……
所以问题是：
如何使用VSTO从Word中的特定合并字段获取值？
浏览 6提问于2014-04-17得票数 1
我已经创建了一个JavaScript函数来检查我的数据库中是否已经存在某个数据。我想知道的是，是否有一种方法可以将JavaScript中的输入字段名作为参数传递
以下是我的代码
function checkDataAvailability(displayid,input_id,fieldname)
{
'use strict';
$(document).ready(function(){
//var x = document.getElementByName(fieldname).elements;
浏览 5提问于2016-06-05得票数 0
我是Regex的新手。我试图扫描几个文本文件，并得到一些基于标签字段和行动的字段名。我使用Powershell和regex = ?smi)\b(field|Action)^)+来获得完整的字段和动作字符串。获得字符串后，立即从字段和操作字符串中获取值，然后调用(?<=;).*，但是，它在字段(“等待期(H)"；”等待期(Hrs.)") { }上失败(错误值)。请注意，值可以是引号，也可以是没有引号，但是它总是以这个)结尾，问题是某些字段名可能有(Hrs.)")。
{
group("Default")
{
Field(&
浏览 4提问于2020-05-18得票数 0
我有一个查询，我使用split_part从特定位置获取值：
split_part(fieldname,'{',12)
结果如下：
200;1;235}
200;1;237}
200;1;216}
我的目标是在第二个;之后获取任何东西，比如(235,237,216)，因为;后面的字符数不是静态的。
这意味着有时它会出现1,2,4甚至5位数，字符的数量会根据值的不同而变化。
怎么做？
浏览 3提问于2014-01-29得票数 1
回答已采纳
我们需要替换flink控制台UI，并使用CLI实用程序启用Flink Web控制台的所有功能，对于一些功能，如启动作业、保存点等，我们使用的是Flink CLI。
我的问题是
是否与Flink控制台具有奇偶性？如果没有，是否有其他方法可以通过Flink控制台(如检查/监视工作的背压等)来完成事情，
我正在寻找一种解决方案，在这种解决方案中，待命工程师可以使用命令行/终端完全监控并在flink上操作，而不需要使用web。
提前谢谢
浏览 3提问于2022-01-03得票数 0
回答已采纳
我有一个List<string>和一些从网格视图中获取的保存的值。我需要的是使用一些if语句来检查列表中的这些值是否为空。
一个简单的for循环遍历网格视图中的所有行，并从右列获取值：
for (int i = 0; i < GridView2.Rows.Count; i++)
{
string tasks = GridView2.Rows[i].Cells[3].Text;
datesList.Add(tasks);
}
下面是一个非常简单的代码示例，我用它来检查2是否在列表中：
if (datesList.Contain
浏览 0提问于2011-12-20得票数 2
回答已采纳
无法使用this.refs获取输入类型的值...如何从输入类型获取值
export class BusinessDetailsForm extends Component {
submitForm(data) {
console.log(this.refs.googleInput.value)
}
}
reder() {
return(
<form onSubmit={this.submitForm}>
<Field type="tex
浏览 87提问于2017-03-31得票数 35
回答已采纳
我在把csv文件拷贝到卡桑德拉。我有下面的csv文件，表如下所示。
创建表UCBAdmissions(
id int PRIMARY KEY,
admit text,
dept text,
freq int,
gender text
)
当我使用从'UCBAdmissions.csv‘中复制UCBAdmissions.csv= '，’和HEADER = TRUE；
输出是
在0.318秒内导入24行。
cqlsh>从UCBAdmissions中选择*；
身份证，承认，部门，女性，性别
(0行)
从“UCBAdmissions”中复制“UCBAd
浏览 3提问于2015-12-15得票数 1
我有一个包含子属性的对象，这些子属性也有子属性等等。
我基本上需要找到检索对象上特定字段的值的最佳方法，因为它是字符串形式的完整分层路径。
例如，如果对象的字段company ( object )具有字段id (String)的字段client (Object)，则此路径将表示为company.client.id。因此，如果给出了我想要在对象上获取值的字段的路径，我该如何着手呢？
干杯。
浏览 0提问于2012-10-21得票数 11
回答已采纳
1回答
当我的制作人把所有的消息都流到Kafka之后，在Flink完成处理之后，我希望能够终止Flink作业，这样它就不会继续运行了，这样我就可以知道Flink什么时候完成了所有的数据处理。我也不能使用批处理，因为我需要Flink运行平行于我的Kafka流。
通常，Flink在一个isEndOfStream类中使用DeserializationSchema方法来查看它是否应该提前结束(返回方法中的true将自动结束作业)。但是，当使用Flink作为来源时，新的KafkaSource类不赞成在反序列化器中使用isEndOfStream方法，不再检查它是否应该结束流。还有其他方法可以提前终止Flink工作
浏览 13提问于2022-10-03得票数 0
1回答
我在当地工作。我使用一个函数在网格中显示单选按钮，然后在renderer中调用该函数。我知道这是错误的，这只是为了向你展示我想要的。
我不知道如何应用单选按钮来处理每一行。
我使用extjs3.4。
抱歉，我的英语和混乱。
这是数组：
var data_metas = [['INGRESE META N°1','','','','','','100%','100%'],
['INGRESE META N°2
浏览 3提问于2017-07-25得票数 0
回答已采纳
我需要为不同的应用程序指定不同的Flink设置。换句话说，每个应用程序都应该使用它的自定义文件flink-conf.yaml运行。正确的方法是什么？
我找到了一些旧的建议来声明FLINK_CONF_DIR，指向具有Flink配置文件的自定义目录(例如：)。但是，官方的Flink文档根本没有提到FLINK_CONF_DIR变量(从Flink 1.13开始)。因此，我有疑问，这种方式是正式推荐和支持的Flink开发人员。
更新1:运行的应用程序详细信息
我正在的纱线上运行Flink。下面是我如何启动应用程序：
"$flink_home/bin/flink" \
run-appl
浏览 23提问于2021-05-05得票数 0
回答已采纳
3回答
我有一个带有几个委托方法的对象DataParameterInfo (DPI)，这些方法用于将数据从DataReader转移到POCO或从POCO中获取值。
示例：
new DataParameterInfo<IBulletinPCN>
{
FieldName = "ChangeProcedure",
ParameterName = "@ChangeProcedure",
EntityName = "ChangeProcedure",
DataType = SqlDbType.NVarChar,
浏览 2提问于2012-05-04得票数 0
回答已采纳
我正在构建一个Access数据库，该数据库从一个网站上的Excel转储中获取交易数据。这些事务都有一个唯一的标识符，但随着事务的处理，状态会发生变化。
以前，如果我们需要分析状态的变化，我们会保留每周的Excel转储，并继续挖掘。我们的“主”数据表是从最新转储运行的40,000行Excel文件，这意味着我们丢失了旧的状态数据。我没有访问后端数据库的权限(如果有？)在存储所有这些信息的位置，仅存储Excel生成的报告。
现在，我正在将我们的数据迁移到Access中，有哪些选项可以捕获特定事务可能经历的不同阶段？
浏览 0提问于2013-04-16得票数 1
回答已采纳
3回答
我试图使用flink写一个csv文件作为地板。我正在使用下面的代码并得到错误。
val parquetFormat = new HadoopOutputFormat[Void, String](new AvroParquetOutputFormat, job)
FileOutputFormat.setOutputPath(job, new Path(outputPath))
我得到以下构建错误。有人能帮忙吗？
类型错配；找到: parquet.avro.AvroParquetOutputFormat必需: org.apache.hadoop.mapreduce.OutputFormatVoi
浏览 5提问于2016-09-09得票数 1
1回答
我正在尝试从Apache收集指标到Prometheus。Flink文档说，我需要在我的flink- that . says中添加以下几行：
metrics.reporter.promgateway.class: org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporter
metrics.reporter.promgateway.host: localhost
metrics.reporter.promgateway.port: 9091
metrics.reporter.promgateway.jobName: myJob
浏览 0提问于2018-11-19得票数 3
回答已采纳
我计划了一个从Apache读取的应用程序，经过(可能耗时的)处理后，将数据保存到数据库中。
我的例子是消息，而不是流，但为了可伸缩性，我正在考虑将其插入Spark或Flink，但无法理解这些扩展:我的应用程序，当Spark/Flink的一部分读取了Kafka的一些数据，然后退出或继续阅读吗？
那么，Spark/Flink将如何决定他们必须生成更多的我的应用程序实例来提高吞吐量呢？
谢谢!
浏览 4提问于2017-09-20得票数 0
回答已采纳
使用flink版本1.13.1
我写了一个自定义的度量报告，但在我的flink中似乎不起作用。启动flink时，JobManager显示警告日志，如下所示：
2021-08-25 14:54:06,243 WARN org.apache.flink.runtime.metrics.ReporterSetup [] - The reporter factory (org.apache.flink.metrics.kafka.KafkaReporterFactory) could not be found for reporter kafka. Available f
浏览 10提问于2021-08-25得票数 2
回答已采纳
2回答
我想在apache flink (flink 1.3)中使用deltatrigger，但我在使用以下代码时遇到了一些问题：
.trigger(DeltaTrigger.of(100, new DeltaFunction[uniqStruct] {
override def getDelta(oldFp: uniqStruct, newFp: uniqStruct): Double = newFp.time - oldFp.time
}, TypeInformation[uniqStruct]))
我有这样的错误：
error: object org.apache.flink.api
浏览 0提问于2017-12-06得票数 0
1回答
我有两个节点，码头已经安装在它们上。此外，我还在码头上安装了Apache。首先，在没有显示任何结果的docker中运行此命令：
/home/flink-1.7.0/bin/mesos-appmaster.sh -Djobmanager.heap.mb=1024 -Djobmanager.rpc.port=6123 -Drest.port=8081 -Dmesos.resourcemanager.tasks.mem=1024 -Dtaskmanager.heap.mb=1024 -Dtaskmanager.numberOfTaskSlots=2 -Dparallelism.default=2
浏览 0提问于2019-03-04得票数 1
我在flink- line .line中添加了以下一行：
env.java.opts：“-Ddy.pros.path=/PATH/TO/PROPS/FILE”
在启动作业管理器(jobmanager.sh启动集群)时，我在日志中看到确实识别了jvm选项。
2017-02-20 12:19:23,536 INFO org.apache.flink.runtime.jobmanager.JobManager - JVM Options:
2017-02-20 12:19:23,536 INFO org.apache.flink.runtime.jobmanage
浏览 6提问于2017-02-20得票数 8
回答已采纳
const renderLists = ({fields}) => {
return <tbody>
{fields.map((member, index) =>
<tr key={index}>
<td>
<Field
name={`${member}.item`}
type="text"
component={renderItems}
浏览 2提问于2016-09-12得票数 2
我有一个方法，它接受一个对象和一个字符串(字符串是对象上一个字段的名称)。因此，我从object类中获取字段，并使用field.get(object)来获取值。但我希望将该值转换为该字段定义为的任何类。有没有一种方法可以在不需要使用@suppressWarning(“未选中”)的情况下进行强制转换？
这是我的代码。我将其缩短为只使用一个对象，但原来的方法使用了一个对象列表，并返回一个map，其中map-key是字段的对象值，map-value是对象。
public static <T1,T2> T2 getFieldValue(T1 obj, String fieldName){
浏览 6提问于2017-02-09得票数 0
回答已采纳
我正在使用Flink，我正在使用Kafka连接器。我从flink收到的消息是一个逗号分隔的项目列表。"'a'，'b'，‘c’，1，0.1....‘12:01:00.000’“其中一个包含事件时间，我想将此事件时间用于每个分区的水印(在kafka源代码中)，然后使用此事件时间进行会话窗口。我的情况与往常有点不同，因为据我所知，人们通常使用"kafka时间戳“和SimpleStringSchema()。在我的例子中，我必须编写自己的反序列化程序来实现DeserializationSchema并返回一个元组或Pojo。所以基本上用我自己的函数替换Si
浏览 69提问于2020-10-24得票数 1
回答已采纳
我们在一个由3台机器组成的集群上使用Flink-1.4。我们使用以下命令在一台机器上启动了JobManager
bin/jobmanager.sh start flip6
接下来，我们使用以下命令在两台机器上启动TaskManager：
bin/taskmanager.sh start flip6
但是，我们没有看到Flink Dashboard Web UI打开。我们在日志中也看不到任何错误。我们是否遗漏了什么，可能是在配置文件中？
这是JobManager的日志：
浏览 9提问于2018-01-15得票数 0
回答已采纳
Flink Web UI有一个出色的背压部分。但我看不到任何指标，由普罗米修斯记者给出，可以用来检测背压的相同方式为Grafana仪表盘。 ? 有没有办法在Flink Web UI之外获得相同的指标？使用此处描述的指标https://ci.apache.org/projects/flink/flink-docs-stable/monitoring/metrics.html。或者甚至有一个普罗米修斯刮刀来抓取web api？
浏览 68提问于2019-03-13得票数 2
回答已采纳
在我的场景中，Flink将度量标准发送到Datadog。Datadog主机映射如下所示：{我不知道为什么在这里显示延迟)
Flink指标被发送到本地主机。这里的问题是
flink-conf.yaml文件配置如下
# adding metrics
metrics.reporters: stsd , dghttp
metrics.reporter.stsd.class: org.apache.flink.metrics.statsd.StatsDReporter
metrics.reporter.stsd.host: localhost
metrics.reporter.sts
浏览 8提问于2017-10-29得票数 0
回答已采纳
我想再问一次，如果有人能帮我把它插入到我的字段专栏中并更新它，这已经困扰了我相当一段时间了。
这张桌子是这样的；
----------------------------------------
id | type | multiplier | rdMultiplier |
----------------------------------------
1 | Reg | null | null |
2 | Spec | null | null |
3 | DReg | null | null
浏览 0提问于2014-03-27得票数 0
回答已采纳
我需要获取这个存储过程的值，并将其作为参数传递给另一个存储过程。
SqlCommand GetName = new SqlCommand("usp_GetName", sqlcon);
GetName.CommandType = CommandType.StoredProcedure;
GetName.Parameters.AddWithValue("@printer", printerguid);
此存储过程有一个简单的select语句，该语句在任何给定点仅返回一行。我需要获取数据并将tat作为参数传递到其他地方...请告诉我如何从此存储过程中获取值并将其存储
浏览 1提问于2012-11-20得票数 0
回答已采纳
Apache Flink 1.10.2的REST API中存在CVE-2020-17518漏洞。现在我想关闭REST API服务，我应该如何设置配置文件？
浏览 2提问于2021-01-14得票数 0
我有一个maven项目，我在其中添加了flink库的依赖项，如下所示：
<repositories>
<repository>
<id>apache.snapshots</id>
<name>Apache Development Snapshot Repository</name>
<url>https://repository.apache.org/content/repositories/snapshots/</url&
浏览 0提问于2016-12-05得票数 1
2回答
我想在django模型中动态获取值 users = User.objects.all()
for user in users:
print(user.first_name) # works fine
# now how to make it dynamic
onefield = "first_name"
print(user.onefield) # come error
# and for list of field n
浏览 49提问于2021-09-22得票数 0
第一次尝试玩flink码头形象。我听从的指示，上面写着
您可以运行JobManager (母版)。
$ docker run --name flink_jobmanager -d -t flink jobmanager
您还可以运行TaskManager (worker)。注意，工作人员需要直接或通过JobManager注册，这样主服务器就开始发送任务给他们执行。
$ docker run --name flink_taskmanager -d -t flink taskmanager
有人能解释一下这个任务管理器是如何从这些命令中向职务经理注册的吗？
谢谢
浏览 3提问于2020-03-20得票数 2
回答已采纳
1回答
问题是在进行案例类提取时，映射函数中存在问题。case类不可序列化。我已经隐式地定义了DefaultFormats格式。
package org.apache.flink.quickstart
import java.util.Properties
import com.fasterxml.jackson.databind.{JsonNode, ObjectMapper}
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import org.apache.flink.api.scala._
import org.apa
浏览 3提问于2017-05-02得票数 1
回答已采纳
我使用Flink v.1.13.2来管理一个工作经理，三个任务经理。
由于某些原因(我无法找出原因)，任务管理器连接正在丢失。下面是我找到的日志：
2022-02-17 21:19:55,891 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Sink: Print to Std. Out (13/32) (f0ff88713cc3ff5ce39e7073468abed4) switched from RUNNING to FAILED on 1.2.3.5:39309-f61daa @ serve
浏览 32提问于2022-02-18得票数 0
首先，我在Kafka topic中加载了一个CSV文件，我可以通过中的Flink程序打印该主题。代码如下： final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
Properties prop = new Properties();
prop.setProperty("bootstrap.servers",
"10.32.0.2:9092,10.32.0.3:9092,10.32.0.4:9092"
浏览 16提问于2019-06-24得票数 0
回答已采纳
1回答
我一直在研究flink。我对flink中的内存管理有疑问。在flink中内存是如何管理的。在flink中是否也有类似的功能，比如"blocks in spark“？请让我知道，如果有好的地方阅读flink中的内存管理。
浏览 12提问于2019-02-15得票数 0
回答已采纳
2回答
我的.rdlc页面中有一个文本框，它从sql server获取值。但是，如果textbox的值为NULL，则希望向该文本框中添加默认值。当前，当来自sql的值为NULL时，它将不显示任何内容。
浏览 5提问于2015-07-17得票数 1
回答已采纳
目前，我正在使用我编写的代码从字段中获取值：
QueryBuilder qb = QueryBuilders.matchPhrasePrefixQuery("tagName", "june");
SearchResponse response = esclient.prepareSearch(index).setTypes(type)
.setQuery(qb)
.execute().actionGet();
SearchHit[]
浏览 0提问于2018-02-12得票数 2
回答已采纳
我在Drupal 7上创建了一个站点。在主题设置中有“站点名”和“站点口号”，但是我至少还需要两个字段来显示标题中的一些信息。我怎么能这么做？(我想留住巴蒂克，我知道选择另一个主题是某种解决方案)
浏览 4提问于2014-07-31得票数 2
回答已采纳
2回答
如何在Flink转换中更新配置？
、、、、
给定一个Flink流作业，该作业将map()操作应用于流。
此map()操作从某些属性读取其配置，并相应地映射数据。例如，配置指定读取属性"input"，并使用不同的属性名"output“将其写入流。这个已经很好了。
现在，配置发生了变化，例如，转换将为输出使用不同的属性名。
因此，我正在寻找一种方式，让所有的Flink任务重新阅读一个新的配置在运行时。
有没有可能
将KafkaSourcewait挂起，直到管道耗尽(flush)trigger集群中的所有任务，以重新读取配置文件(coordinated)resume (coordinated)resume KafkaS
浏览 13提问于2022-10-11得票数 0
我想在我的Flink工作中扼杀卡夫卡的消费者。
通过查看Flink 1.12的源代码，我发现了FlinkConnectorRateLimiter和GuavaFlinkConnectorRateLimiter。但我找不到任何与FlinkKafkaConsumer的速率限制器有关的东西。
如何在Flink 1.12中实现卡夫卡的速率限制？
浏览 0提问于2020-12-10得票数 5
回答已采纳
3回答
我有作为Java类的资源包，用于从数据库中读取值。当我更新db时，我需要重新加载捆绑包，但我不知道怎么做。有人能帮上忙吗？
package model.helpers;
public class Messages_en extends ListResourceBundle {
protected Object[][] getContents() {
// from DB
// ...
}
}
在视图中，我使用如下捆绑包：
<f:loadBundle basename="model.helpers.Messag
浏览 0提问于2010-12-01得票数 10
回答已采纳
我正在尝试在一个简单的Flink管道中使用TestContainers测试Cassandra Sink，该管道使用DataStreamTestBase进行测试： public class CassandraPojoSinkExampleTest extends DataStreamTestBase {
@Rule
public CassandraContainer cassandra = new CassandraContainer<>().withInitScript("testInit3.cql")
.withCrea
浏览 36提问于2020-08-31得票数 0
回答已采纳
尝试按照Flink Kubernetes说明here在minikube上运行word count example，但作业从未完成。Python Beam SDK worker pooler似乎不做任何工作。 除了配置Flink Kubernetes集群的说明之外，我还在taskmanager部署中添加了一个Python SDK工作池。如果我理解正确的话，worker池的目的是执行管道的Python部分。请参阅完整的k8s部署。 apiVersion: apps/v1
kind: Deployment
metadata:
name: flink-taskmanager
namespac
浏览 39提问于2020-09-29得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
