1回答
我使用JsonRowSerializationSchema将Flink的行序列化为JSON。I SQL时间戳序列化存在时区问题。
val row = new Row(1)
row.setField(0, new Timestamp(0))
val tableSchema = TableSchema
.builder
.field("c", DataTypes.TIMESTAMP(3).bridgedTo(classOf[Timestamp]))
.build
val serializer = JsonRowSerializationS
浏览 7提问于2020-11-17得票数 1
作为flink文档中的示例：
input
.keyBy(<key selector>)
.window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)))
.<windowed transformation>(<window function>);
我想为每个键设置不同的偏移量，因为它有不同的时区。我怎么能这么做？
浏览 0提问于2018-05-23得票数 0
回答已采纳
2回答
我想验证一下我的时间戳。我附上了代码片段：
import org.apache.flink.streaming.api.windowing.time.Time;
Date date = new Date(Time.seconds(30).toMilliseconds());
DateFormat formatter = new SimpleDateFormat("HH:mm:ss:SSS");
String dateFormatted = formatter.format(date);
logger.debug(Time.seconds(30).toMillisecon
浏览 15提问于2017-06-29得票数 2
2回答
我正试图在PyFlink中的一个窗口上进行一些聚合。但是，我尝试它时会得到A group window expects a time attribute for grouping in a stream environment.错误。我在窗口定义和select中都有一个time属性。
我做错了什么？
from pyflink.common import Row
from pyflink.table import EnvironmentSettings, TableEnvironment, DataTypes
from pyflink.table.expressions import col
浏览 15提问于2021-06-17得票数 1
我正在为一个使用MySQL作为其数据存储的应用程序构建分析功能。我们有一个基于微服务的架构，也使用Kafka。我们的用例并不真正需要“实时”分析，但这可能会在以后添加。
对于我的用例，我想使用Tableau作为可视化平台，其中的报告将直接嵌入到web应用程序中。
对于卷和用例，我不认为需要基于Hadoop的系统，但Kafka Connect，Spark和Flink是可能的。
我打算构建一个基于星型模式的报告数据库，独立于主生产数据库，由维度和事实表组成，并允许Tableau报告这一点。
我的微服务将使用Avro模式注册表将事件推送到相关主题，然后报告微服务将使用这些事件并更新星型模式。
现在我
浏览 0提问于2018-03-23得票数 3
我正试图遵循的官方教程。
安装先决条件之后，我执行以下命令：
$ mvn archetype:generate \
-DarchetypeGroupId=org.apache.flink \
-DarchetypeArtifactId=flink-walkthrough-datastream-java \
-DarchetypeVersion=1.15.0 \
-DgroupId=frauddetection \
-DartifactId=frauddetection \
-Dversion=0.1 \
-Dpackage=spendre
浏览 7提问于2022-06-24得票数 1
回答已采纳
1回答
Flink码头图像
、、、
我是新来的码头，我正在努力创建Flink码头形象。码头形象正在显示成功创建。但是我有一个疑问，我不能像在Dockerfile中提到的那样在/opt路径中看到Flink二进制文件。
如何知道我的Flink码头形象是成功创建的。
屏幕截图从控制台，很少一组命令是突出显示红色，意味着它可以吗？
创建显示Flink码头图像的状态：
请帮帮我。谢谢你..。
全日志：
sudo /home/develk/cntx_eng/build.sh \
> --job-artifacts /home/develk/cntx_eng/FlinkContextEnginePoc-0.0.1-S
浏览 2提问于2019-12-09得票数 0
回答已采纳
需要帮助！谁能给我指引一条正确的路？
以下是我的代码和日志的片段。
DataStream<ObjectNode> stream = env.addSource(KafkaConsumer.getKafkaConsumer());
DataStream<MyDataObject> dataStream = stream.flatMap(new DataTransformation());
我使用flatMapFunction来处理我的输入对象并获得多个对象。
以下是stackTrace：
java.lang.RuntimeException: Buffer pool is
浏览 8提问于2018-01-16得票数 4
我以独立模式在k8s上部署了几个flink，并通过一个promethus-pushgateway导出它们的指标。
问题是：
度量数据间歇性地到达promethus，当在grafana中显示时，会导致点间的间隙。
目标：
monitoring/pushgateway/0 (1/1 up)
Endpoint: http://172.19.88.111:9091/metrics
State : UP
Labels: endpoint="tcp" instance="172.19.88.111:9091" job="pushgateway" na
浏览 8提问于2021-05-24得票数 0
2回答
我正在尝试创建一个sbt Scala项目，添加一些来自Flink的代码。我已经成功地添加了除Flink-ML以外的大多数Flink项目。
我尝试添加不同的依赖项，错误如下：
sbt.build配置：
resolvers += Resolver.mavenLocal
// I have downloaded the Flink project and compiled in local.
libraryDependencies += "org.apache.flink" % "flink-ml_2.10" % "1.0-SNAPSHOT"
错误如下
浏览 2提问于2016-02-23得票数 1
Added Depedency Pom Details :
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.11</artifactId>
<version>1.7.1</version>
</dependency>
<depe
浏览 0提问于2019-04-04得票数 2
回答已采纳
我想在我的工作中添加一个JVM选项。因此，我在flink- line . the中添加了以下一行：
env.java.opts: "-Dspring.config.location=/root/application.properties"
当我使用命令./start-local.sh启动flink时，jobmanager和taskmanager很好地加载了JVM选项：
JobManager日志：
org.apache.flink.runtime.jobmanager.JobManager - JVM Options:
org.a
浏览 0提问于2018-03-01得票数 0
回答已采纳
我需要将表结果保存到S3上的orc中，我就是这样做的：
tEnv.createTemporaryTable("my_output_table", TableDescriptor.forConnector("filesystem")
.schema(outputSchema)
.option("path", s3OutputPath)
.format(FormatDescriptor.forFormat("orc").build())
浏览 15提问于2022-10-17得票数 0
回答已采纳
我正在尝试创建到oracle数据库的Flink JBDC接收器。当在本地运行时(从junit测试和minicluster)，它可以工作，但是当部署在k8s中时，它会抛出一个异常，说它找不到合适的驱动程序。Classpath是：
Classpath: /flink/lib/flink-cep-scala_2.12-1.13.5-stream1.jar:/flink/lib/flink-connector-jdbc_2.12-1.13.5.jar:/flink/lib/flink-csv-1.13.5-stream1.jar:/flink/lib/flink-json-1.13.5-stream1
浏览 16提问于2022-05-24得票数 1
回答已采纳
我有一个容器化的flink集群，具有独立的JobManager和2个任务管理器。当我提交一个包含我需要的连接器和代码的胖jar时，一切都会正常运行，但是一个只有我的代码的jar将失败--尽管我已经将连接器jar添加到/opt/flink/lib，容器进程显示它们将被添加到类路径中。当我试图从IDE进行远程连接时，也会发生同样的失败；不用说，每次都要打包一个胖jar，这会带来糟糕的开发体验。
我需要做什么才能让flink理解我放置在/opt/flink/lib中的额外jars应该与所有新作业一起使用？
类加载程序异常：
org.apache.flink.streaming.runtime.ta
浏览 0提问于2019-04-12得票数 1
回答已采纳
0回答
所以我用它安装了一个样例Flink项目。
我正在尝试使用这个模板，它让我可以开始编写一个Flink摄取应用程序，而不必担心依赖关系，但它适得其反。当我尝试与sbt同步时，我发现找不到Flink (不是要说谎，但这部分对我来说甚至是模糊的)。我想知道是否有人知道如何让我的项目找到Flink。使用这项技术非常令人兴奋。
Error while importing sbt project:
Java HotSpot(TM) 64-Bit Server VM warning: ignoring option MaxPermSize=384M; support was removed in 8.0
[
浏览 6提问于2017-12-05得票数 0
回答已采纳
我试图阅读卡夫卡主题的数据，并成功地阅读了它。但是，我希望提取数据并将其作为Tuple返回。因此，我尝试执行map操作，但它不允许我通过说cannot resolve overloaded method 'map'来执行。下面是我的代码：
package KafkaAsSource
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.datastream.DataStream
import org.apache.f
浏览 6提问于2021-12-27得票数 0
回答已采纳
我正在尝试使用我的felipeogutierrez/explore-flink:1.11.1-scala_2.12镜像可用here到kubernetes集群配置中，就像它说here一样。我用maven编译了我的项目https://github.com/felipegutierrez/explore-flink，并用这个Dockerfile扩展了默认的flink图像flink:1.11.1-scala_2.12 FROM maven:3.6-jdk-8-slim AS builder
# get explore-flink job and compile it
COPY ./java/expl
浏览 89提问于2020-09-21得票数 1
回答已采纳
我刚开始使用flink，我正在实现一个模式识别模块(不使用CEP实现模式匹配)，它将从EventHub主题读取json流，如果模式匹配，则推送到另一个事件中心主题。我的模块功能如下
从接收JSON负载
I使用的是一个RichSourceFunction，它将从API读取模式并发送到广播流
我正在使用Flink BroadcastProcessFunction来根据广播状态中可用的模式列表来处理数据，并且我没有使用键流或我的程序中的任何状态，因为没有状态的范围，我只需要检查某些值是否存在于JSON .中。
我读到flink在内部维护状态，即使没有状态显式实现
下面的设置为检查点，我平均每小时收到
浏览 8提问于2021-01-29得票数 0
2回答
在电子病历上对S3进行闪烁写入
、、、、
我正在尝试写一些输出到S3使用电子病历与Flink。我使用的是Scala 2.11.7、Flink 1.3.2和EMR 5.11。但是，我得到了以下错误：
java.lang.NoSuchMethodError: org.apache.hadoop.conf.Configuration.addResource(Lorg/apache/hadoop/conf/Configuration;)V
at com.amazon.ws.emr.hadoop.fs.EmrFileSystem.initialize(EmrFileSystem.java:93)
at org.apache.fl
浏览 3提问于2018-01-23得票数 1
1回答
我正在尝试运行，但我不想将数据转移到Elasticsearch，我想使用InfluxDB。
注意:演练中的代码(上面的链接)正在按预期工作。
为了使这个工作，我们需要把InfluxDB连接器内的码头容器。其他Flink连接器被放置在容器中，在中使用以下命令
# Download connector libraries
RUN wget -P /opt/flink/lib/ https://repo.maven.apache.org/maven2/org/apache/flink/flink-json/${FLINK_VERSION}/flink-json-${FLINK_VERSION}.ja
浏览 3提问于2022-06-13得票数 1
回答已采纳
我是新的Flink CEP，并试图测试基本的东西-在下面的代码，我的期望是所有的输入应该匹配模式，并应打印为匹配的结果。但不知何故，没有什么比('matechedStream.print()')更能说明原因吗？
任何建议/帮助都将不胜感激。
package com.o9.flink;
import com.o9.flink.asyncio.DemandSupply;
import org.apache.flink.cep.CEP;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
浏览 9提问于2022-08-09得票数 0
回答已采纳
我目前正在使用Flink 1.7 +gcs-连接器库。我试图让StreamingFileSink写到GCS桶中，并遇到以下异常：
我遇到了一个Jira： --但我不清楚代码是否曾经合并过。
在需要做什么方面有什么帮助是非常感谢的，以使这一工作？
java.lang.UnsupportedOperationException: Recoverable writers on Hadoop are only supported for HDFS and for Hadoop version 2.7 or newer
at org.apache.flink.runtime.fs.hdfs.Ha
浏览 3提问于2019-12-03得票数 1
回答已采纳
我有一个scala文件，包含这个导入和这个代码。我不知道如何解决这个重载构造函数的问题：
package uimp
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.util.serialization.
{SimpleStringSchema,JSONKeyValueDeserializationSchema}
import org.apache.flink.core._
浏览 4提问于2020-06-02得票数 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
回答已采纳
我按照的说明将flink指标导出到普罗米修斯，但是默认情况下它似乎只将与作业管理器相关的指标导出到普罗米修斯，如下所示：
打开，我只看到以下信息，没有找到任务或任务管理器相关的指标。
# HELP flink_jobmanager_Status_JVM_Memory_Mapped_MemoryUsed MemoryUsed (scope: jobmanager_Status_JVM_Memory_Mapped)
# TYPE flink_jobmanager_Status_JVM_Memory_Mapped_MemoryUsed gauge
flink_jobmanager_Status
浏览 22提问于2018-08-22得票数 2
回答已采纳
1回答
我有一个flink工作，它消耗kafka的数据，做一些无状态的平面图，并产生数据到kafka，这是一个非常低的工作量的工作。
它通常会毫无问题地获取检查点，直到作业需要从检查点恢复为止，例如，它只是无法恢复具有下面堆栈跟踪的状态。
状态很小，我相信它只是卡夫卡的偏移量，它是用AT_LEAST_ONCE语义运行的。
所有的运算符都设置了.uid()，我完全没有想法。
这是尝试从检查点重新启动时出现的错误：
java.lang.Exception: Exception while creating StreamOperatorStateContext.
at org.apache.flin
浏览 3提问于2021-06-10得票数 0
我尝试在Flink (版本1.14.0)上使用可查询状态，但不幸的是，我一直收到以下错误： 2021-11-07 11:10:55,795 ERROR org.apache.flink.queryablestate.network.AbstractServerHandler [] - Error while handling request with ID 1
java.lang.OutOfMemoryError: Direct buffer memory
at java.nio.Bits.reserveMemory(Bits.java:175) ~[?:?]
at java
浏览 31提问于2021-11-07得票数 0
我使用intellij、maven 3和flink 1.15.1编写有状态流作业。我正在尝试为我的自定义KeyedProcessFunction编写单元测试，并尝试遵循文档以及添加提到的的依赖项。我对使用KeyedOneInputStreamOperatorTestHarness类很感兴趣，但是在我的任何依赖项中都找不到这个类(下面发布的修改后的pom )。我唯一能找到这门课的方法就是
<dependency>
<groupId>org.apache.flink</groupId>
<ar
浏览 8提问于2022-08-01得票数 1
回答已采纳
1回答
当我将Flink Java应用程序从1.12.2升级到1.12.3时，我会得到一个新的运行时错误。我可以脱下我的Flink应用程序到这两条线上：
public class TableEnvOnly {
public static void main(String[] args) throws Exception {
final StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment
浏览 8提问于2021-04-29得票数 0
回答已采纳
我正在设置一个使用ZooKeeper和3个TaskManagers的JobManager HA集群。我在kubernetes中这样做，并且需要JM通过全主机名(hostname -f而不是hostname)来寻址to，因为它们是在有状态集中。 目前，似乎TMs将他们的地址报告给JM只是作为nostname… 2019-04-12 08:58:32,426 WARN akka.remote.ReliableDeliverySupervisor - Association with remote system [akka.tcp://flink-m
浏览 141提问于2019-04-13得票数 0
在Flink SQL上使用带有Kerberos身份验证的Kafka数据源，Flink的本地测试已经通过，但当我将任务推到纱线上时，错误消息提示无法找到Kafka的JAAS文件！
与Flink SQL相关的代码如下：
create table source_sensor(
id VARCHAR,
ts bigint,
vc double)
WITH (
'connector' = 'kafka',
'topic' = 'huangfu_0110'
浏览 23提问于2022-01-24得票数 1
我试图在Dockerfile中对Flink应用程序进行，但看起来忽略了它。
Dockerfile如下：
FROM flink:1.14.2-scala_2.12-java11
ENV FLINK_HOME=/opt/flink
COPY custom-docker-entrypoint.sh /
RUN chmod a+x /custom-docker-entrypoint.sh
COPY --chown=flink:flink --from=build /target/*.jar /opt/flink/flink-web-upload/
ENTRYPOINT ["/custom
浏览 8提问于2022-09-05得票数 1
回答已采纳
1回答
嗨，我正在尝试运行一个flink scala应用程序，它从kafka读取，应用一些查找转换，然后写到kafka。
Flink版本1.12.1
我在当地测试过，效果很好。但是，当我尝试使用原生kubernetes集成在集群上运行它时，我会看到下面这样的奇怪错误。
集群看起来也很好，因为我试图在集群上运行一个wordcount应用程序，它运行得很好。
异常不清楚，堆栈跟踪也显示了任务管理器堆栈跟踪，因此不知道应用程序中的问题可能在哪里。这可能是序列化问题吗？是否有一种方法来调试这些问题，并在应用程序代码中找到有问题的实际点？
org.apache.flink.streaming.runtime.t
浏览 2提问于2021-09-01得票数 0
回答已采纳
2回答
我已经安装了Flink，Scala和sbt Flink版本: 1.9.1 Scala版本: 2.10.6 Sbt版本: 1.3.7
我在build.sbt中做了相关的更改。编译命令失败，这里是相关信息。我们非常感谢您提供的任何信息
**版本信息
osboxes@osboxes本地$ scala -version
Scala代码运行器版本2.10.6 --版权所有2002-2013，LAMP/EPFL
osboxes@osboxes $ flink --版本
版本: 1.9.1，提交ID: 4d56de8
osboxes@osboxes $ sbt -version
本项目sbt版本: 1.3.
浏览 1提问于2020-01-30得票数 1
使用Flink 1.7.1为kubernetes flink上的单个作业集群构建它无法加载核心站点xml，尽管在类路径上，导致忽略配置，但是，如果我将ENV变量放入AWS_SECRET_ACCESS_KEY AWS_ACCESS_KEY_ID，它通过找到它来工作，但如果我依赖于core-site.xml，它永远不会在没有环境变量的情况下工作。 我目前正在复制core-site.xml，就像Dockerfile文件中显示的那样，并且正如文档中所说的那样，将HADOOP_CONF_DIR作为指向它的环境变量。它仍然不会加载它，从而导致NoCredentialsProvider。 例外情况是： C
浏览 128提问于2019-01-25得票数 1
下面提到的Apache测试代码在我想使用KeyedOneInputStreamOperatorTestHarnessK、IN、Int或Long作为key (K)参数时不会在Scala中编译。可以使用字符串、整数或java.lang.Long类型的泛型参数编译相同的代码：
import org.apache.flink.api.java.functions.KeySelector
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.functions.KeyedProces
浏览 7提问于2021-12-31得票数 2
我目前有一个运行在独立Kubernetes (v1.16)中的Flink (1.12)集群。出于我们的目的，我们采用了application cluster模式部署。 为了使我们的flink集群对故障更有弹性，我们希望将HA添加到我们当前的设置中，我已经阅读了文档，并遵循了为我们的给定设置(here)推荐的示例配置。 flink-conf.yaml jobmanager.rpc.address: {{ $fullName }}-jobmanager
jobmanager.rpc.port: 6123
jobmanager.memory.process.size: 1600m
浏览 77提问于2021-08-24得票数 0
我正在为集成hdfs到flink而苦苦挣扎。
Scala二进制版本: 2.12，
Flink (集群)版本: 1.10.1
这个配置和HADOOP_CONF_DIR在任务管理器中也是一样的。
pom.xml；
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId&
浏览 2提问于2020-06-24得票数 0
我在kubernetes配置映射中配置Apache flink(v1.10.0) jar存储路径，如下所示： web.upload.dir: file:///opt/flink/data/jars 但是当从taskmanager上传jar文件时，抛出这个错误： 2020-05-03 09:20:21,444 WARN org.apache.flink.runtime.rest.FileUploadHandler - File upload failed.
java.lang.IllegalArgumentException: UploadDirectory is
浏览 64提问于2020-05-03得票数 1
回答已采纳
我正在从AWS控制台(使用Add )启动EMR中的Flink作业，并希望传递JVM参数。根据Flink文档，我们应该按照下面的方式传递args。
flink run -m yarn-cluster -p 2 /home/hadoop/flink-0.0.1-SNAPSHOT.jar -yD env.java.opts="-Dspring.profiles.active=dev -Djob.name=Flink_Job_1 -Dkafka.brokers:<>:9092" -yid application_1591030260237_0047
我有两个问题
flink
浏览 18提问于2020-06-23得票数 0
我有一个flink JobManager，只有一个TaskManager运行在Kubernetes之上。为此，我使用Service和Deployment来表示TaskManager和replicas: 1。
apiVersion: v1
kind: Service
metadata:
name: flink-taskmanager
spec:
type: ClusterIP
ports:
- name: prometheus
port: 9250
selector:
app: flink
component: taskmanager
the Depl
浏览 5提问于2020-09-24得票数 0
回答已采纳
下面是SBT配置
// https://mvnrepository.com/artifact/org.apache.flink/flink-scala
libraryDependencies += "org.apache.flink" %% "flink-scala" % "1.9.1"
// https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-scala
libraryDependencies += "org.apache.flink" %%
浏览 1提问于2020-02-27得票数 1
回答已采纳
1回答
我使用了flink 1.11，但任务管理器找不到org.apache.flink.metrics.influxdb.InfluxdbReporterFactory WARN org.apache.flink.runtime.metrics.ReporterSetup [] - The reporter factory (org.apache.flink.metrics.influxdb.InfluxdbReporterFactory) could not be found for reporter influxdb. Available factories: [
浏览 47提问于2020-07-17得票数 0
回答已采纳
我使用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
在我的本地环境中，当我向Flink (v1.0.3)提交一个新作业时，我得到了这个错误。
原因: org/apache/flink/streaming/runtime/operators/CheckpointCommitter at org.apache.flink.streaming.connectors.cassandra.CassandraSink.addSink(CassandraSink.java:164) at com.xxx.yyy.sample.backend.flink.AAAAA.main(AAAAA.java:99) at sun.reflect.NativeMethod
浏览 10提问于2016-07-28得票数 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
回答已采纳
我正在运行一个具有以下配置的flink docker镜像。 version: '2.1'
services:
jobmanager:
build: .
image: flink
volumes:
- .:/usr/local/lib/python3.7/site-packages/pyflink/lib
hostname: "jobmanager"
expose:
- "6123"
ports:
- "8081:8081"
comma
浏览 121提问于2020-10-11得票数 1
获取以下flink作业提交错误， @centos1 flink-1.10.0]$ ./bin/flink run -m 10.0.2.4:8081 ./examples/batch/WordCount.jar --input file:///storage/flink-1.10.0/test.txt --output file:///storage/flink-1.10.0/wordcount_out
Job has been submitted with JobID 33d489aee848401e08c425b053c854f9
----------------------------
浏览 1218提问于2020-03-19得票数 0
回答已采纳
我正在尝试使用Scala和Flink将消息发布到Kafka主题中。但是，当使用中提供的代码创建FlinkKafkaProducer对象时，它告诉我无法应用构造器。以下是代码示例：
val studentProducer = new FlinkKafkaProducer[String](
"my_topic", // target topic
new SimpleStringSchema(), // serialization schema
properties, // producer config
浏览 103提问于2021-05-24得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
