接收器可以成为Apache Flink中的下一个数据源吗
接收器（Sink）是Apache Flink中的一种数据输出组件，用于将数据流发送到外部系统或存储介质。它可以将数据流写入各种目标，如文件系统、数据库、消息队列等。
在Apache Flink中，接收器主要用于将处理后的数据流输出到外部系统，而不是作为数据源。数据源（Source）是用于从外部系统或存储介质读取数据流的组件。
因此，接收器不能成为Apache Flink中的下一个数据源。接收器的作用是将数据流输出到外部系统，而数据源的作用是从外部系统读取数据流。
如果您需要将数据流作为数据源输入到Apache Flink中，可以使用Flink提供的各种数据源组件，如Kafka数据源、Socket数据源、文件数据源等。这些数据源组件可以将外部系统的数据流作为输入，供Apache Flink进行处理和分析。
关于Apache Flink的更多信息和相关产品介绍，您可以参考腾讯云的官方文档：
- Apache Flink产品介绍：https://cloud.tencent.com/product/flink
- Apache Flink文档：https://ci.apache.org/projects/flink/flink-docs-release-1.14/
相关·内容
我正在尝试一个简单的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
回答已采纳
我目前正在使用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
回答已采纳
作为一个玩具示例，假设我希望构建一个web应用程序，该应用程序接收带有ID和$ payed的HTTP请求，并以ID在24小时窗口中支付的总额返回响应。使用Apache，如果输入和输出是Kafka流，当然可以创建该应用程序。
，但是否支持/可能创建一个Flink应用程序，其中输入是HTTP请求，输出是对HTTP请求的响应？
如果Flink不支持这一点，是否还会有另一个数据流框架支持这一点？
(对于上面的玩具示例，我想有比使用Flink更简单的解决方案，但我的实际使用是通过HTTP“流”请求的许多多个窗口和其他有状态的计算。)
浏览 4提问于2020-06-16得票数 0
回答已采纳
我的flink应用程序在启动时抛出这样的异常：
Exception in thread "main" org.apache.flink.api.common.InvalidProgramException: org.apache.ibatis.binding.MapperProxy@3fe8ad3f is not serializable. The object probably contains or references non serializable fields.
at org.apache.flink.api.java.ClosureCleaner.clea
浏览 8提问于2021-03-21得票数 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
回答已采纳
我试图用Flink编写一个计算，它需要两个阶段。
在第一个阶段，我创建一个图并获取它的顶点ids：
List<String> ids = graph.getVertexIds().collect();
在第二阶段，我希望使用这些ids来运行每个顶点的SingleSourceShortestPath。
for (String id: ids){
System.out.println("Source Id: "+id);
graph.run( new SingleSourceShortestPaths<String, String&
浏览 6提问于2017-11-28得票数 0
回答已采纳
我正在尝试建立一个非常基本的flink工作。当我试图运行时，得到以下错误：
Caused by: java.lang.IllegalStateException: No operators defined in streaming topology. Cannot execute.
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getStreamGraph(StreamExecutionEnvironment.java:1535)
at org.apache.flink.stre
浏览 1提问于2019-03-04得票数 11
回答已采纳
我阅读了使用RichSinkFunction比直接调用DB方法的优点。因此，我决定写我自己的RichSinkFunction。
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import com.datastax.driver.core.{Session, Cluster}
class CassandraAsSink extends RichSinkFunction
浏览 4提问于2022-01-05得票数 1
如何识别任务中产生背压的根本原因？(即多个操作符的哪个操作符会导致背压)
有相关的日志吗？(失败的跟踪StackTraceSampleCoordinator -“接收到的后期堆栈跟踪示例”没有出现在任何日志中)
还有其他我能用的工具吗？
=====================================
下面是我所封装的内容:在Flink作业执行过程中，会显示一个背压指示。据我所知，引起的任务是接续具有BP指示的“最新”任务的任务。此任务正在运行多个操作符的流:还原、映射和接收器。分析作业度量没有帮助--从前面的操作符中得到的是在这个操作符中得到了什么。下列工作计划的第
浏览 0提问于2018-02-19得票数 2
我需要在Flink流媒体方面的帮助。下面我制作了一个简单的Hello类型的代码。这会从RabbitMQ中流Avro消息并将其保存到HDFS。我希望有人能检查代码，也许它能帮助其他人。
我为Flink流找到的大多数示例都会将结果发送给std-out。实际上我想把数据保存到Hadoop。我读过，理论上，你可以和Flink一起流到任何你喜欢的地方。实际上，我还没有找到任何将数据保存到HDFS的例子。但是，根据我找到的例子，以及试验和错误，我附带了以下代码。
这里的数据来源是RabbitMQ。我使用客户端应用程序将"MyAvroObjects“发送到RabbitMQ。MyAvroObject.
浏览 8提问于2015-12-28得票数 4
我们有一个flink流作业，它读取卡夫卡的数据并将其汇到S3。我们使用flink的内部流文件接收器API来实现这一点。然而，几天后，作业失败了，无法从失败中恢复。消息说它无法从找到文件。我们想知道可能的根本原因是什么，因为我们真的不想丢失任何数据。
谢谢。
整个输出如下所示
java.io.FileNotFoundException: No such file or directory: s3://bucket_name/_part-0-282_tmp_b9777494-d73b-4141-a4cf-b8912019160e
at org.apache.flink.fs.shaded.
浏览 1提问于2019-11-27得票数 1
我想使用flink sql client创建一个配置单元表。 我可以成功创建表t2，但是当我查询t2时，它会报错 Table options do not contain an option key 'connector' for discovering a connector. 我已经在conf/sql-client-defaults.yaml文件中设置执行类型为batch。 我会问这里有什么问题。谢谢! Flink SQL> use testdb1;
Flink SQL> create table t2(id int,name string);
[INFO
浏览 103提问于2020-12-21得票数 0
回答已采纳
1回答
我能在Flink中为任务的不同部分设置不同程度的并行性吗？例如，Flink如何解释以下示例代码？两个自定义从业者MyPartitioner1，MyPartitioner2，将输入数据划分为两个4和2个分区。
partitionedData1 = inputData1
.partitionCustom(new MyPartitioner1(), 1);
env.setParallelism(4);
DataSet<Tuple2<Integer, Integer>> output1 = partitionedData1
.mapPartition(new calcu
浏览 2提问于2015-12-02得票数 9
回答已采纳
我有一个在AWS EMR中运行的高度并行(400)的Flink应用程序。它使用BucketingSink源码Kafka并汇入S3 (使用RocksDb后端设置检查点)。目的地使用"s3a://“前缀定义。Flink作业是一个持续运行的流媒体应用程序。在任何给定的时间，所有工作进程加在一起都有可能生成/写入400个文件(由于400个并行度)。几天后，其中一个工人将失败，并出现以下异常：
org.apache.hadoop.fs.s3a.AWSS3IOException: copyFile(bucket/2018-09-01/05/_file-10-1.gz.in-progress, bu
浏览 4提问于2018-12-05得票数 0
我正在试用Flink版本1.1.3中的ALS代码，使用：
mvn archetype:generate \
-DarchetypeGroupId=org.apache.flink \
-DarchetypeArtifactId=flink-quickstart-scala \
-DarchetypeVersion=1.1.3 \
-DgroupId=org.apache.flink.quickstart \
-D
浏览 4提问于2017-01-26得票数 0
回答已采纳
我正在使用Apache进行流处理。
在订阅了来自源(ex:Kafka、AWS Kinesis数据流)的消息，然后使用Flink运算符对流数据应用转换、聚合等之后，我希望缓冲最终消息(计数为ex:1000)，并在单个请求中将每个批发送到外部REST。
如何在Apache中实现缓冲机制(将每1000条记录作为一个批处理创建)？
Flink pipileine:流源->转换/减少使用操作符->缓冲区1000条消息-> post到REST
感谢你的帮助！
浏览 4提问于2019-11-07得票数 1
回答已采纳
1回答
我可以从IntelliJ运行我的Flink作业没有问题，但是当我尝试在flink standalone上运行它时……
wget ... flink-1.4.0-bin-hadoop27-scala_2.11.tgz
tar xf flink-1.4.0-bin-hadoop27-scala_2.11.tgz
./flink-1.4.0/bin/start-local.sh
./flink-1.4.0/bin/flink run ../mypath/target/messagehub-to-s3-1.0-SNAPSHOT.jar ...
我得到了错误消息：
Caused by: java.lan
浏览 3提问于2018-01-24得票数 0
我刚刚使用maven原型(即模板)创建了新的Flink项目(查看文章)。
mvn archetype:generate \
-DarchetypeGroupId=org.apache.flink \
-DarchetypeArtifactId=flink-quickstart-java
1.14.4是目前的版本。该项目可以编译。但是当运行BatchJob或StreamingJob时
public class BatchJob {
public static void main
浏览 39提问于2022-04-22得票数 0
回答已采纳
我试图理解Apache仪表板显示的"Bytes接收/发送“意味着什么。对于某些上下文，CSV文件托管在HDFS服务器上，我正在将结果写入我的计算机上的本地TXT文件。Flink也在我的机器上本地运行。考虑到这一点，"Bytes server“似乎意味着"Bytes从HDFS服务器发送到我的机器”，而"Bytes接收“似乎意味着"Bytes从我的机器发送到HDFS服务器”。，这是正确的解释吗？
时间线显示的重叠任务也让我有点困惑。奇怪的是，连接在过滤前两个数据集完成之前就开始了。，这是预期的行为吗?如果是的话，为什么？
下面是我关于正在
浏览 1提问于2019-05-19得票数 1
回答已采纳
问题：
我想用pyflink读取kafka msg运行逗号./bin/flink run -m 10.0.24.13:8081 -py /usr/local/project/cdn_flink/cdn_demo.py，显示错误：
File "/usr/local/flink-1.14.2/opt/python/pyflink.zip/pyflink/table/table.py", line 1108, in execute_insert
File "/usr/local/flink-1.14.2/opt/python/py4j-0.10.8.1-src.zip/
浏览 48提问于2022-01-14得票数 0
回答已采纳
我试图从cassandra表中提取数据作为数据集，但我遇到了两个问题。
首先，cassandraInputFormat只返回一个元组，我更希望没有tuple12，只需要使用pojo来定义它期望返回的内容。因此，我不知道这是否只是我必须接受的东西，是否有一种方法可以使用pojo代替cassandraConnector ()，或者使用cassandraInputFormat不是获取数据的最佳方式。
另一个问题是，即使我从cassandraInputFormat中提取的数据(不管是元组还是非元组)，我也不知道如何将它设置为数据源。对于文件、csv和HDFS，有很多方法()，但没有一个方法显式地用于c
浏览 4提问于2017-05-11得票数 0
回答已采纳
我试图创建一个简单的应用程序，应用程序将使用Kafka消息，做一些cql转换并发布到Kafka，下面是代码：
JAVA: 1.8 Flink: 1.13 Scala: 2.11 flink-siddhi: 2.11-0.2.2-快照
我使用的是库：
卡夫卡的输入json：
{
"awsS3":{
"ResourceType":"aws.S3",
"Details":{
"Name":"crossplane-test",
"Cr
浏览 6提问于2021-05-23得票数 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
我使用elasticsearch 6.2.2作为flink中的接收器，但在阅读apache flink文档时，我可以看到用于5.x版的连接器，但没有提到6.x版的连接器。
flink是否支持elasticsearch 6.x的任何连接器？或者，我是否应该创建任何http客户端，并在每次创建或更新新文档时发出post请求？或者他们有其他的技术？
浏览 14提问于2018-02-28得票数 0
1回答
Flink表的状态后端
、、、
我可以检查点Apache Flink中的动态表并将其作为持久后端存储到RocksDB中吗？如果是这样，我可以在这里使用20+ GB吗？
浏览 21提问于2021-07-26得票数 0
回答已采纳
我正在尝试运行一个PyFlink作业，它从源Kafka主题获取数据，并将其放入hdfs中。有一个奇怪的与SQL相关的错误不断出现。这来自Apache (PyFlink) Table中的SQL语句：
SQL:
sql_statement_sink = """
CREATE TABLE avro_sink (
timeTime STRING,
correlationId STRING,
spanId STRING,
appN
浏览 6提问于2021-11-21得票数 0
我正在尝试执行Flink (1.12.1)批处理作业，其步骤如下：
自定义SourceFunction与MongoDB连接
做任何平面图和地图来转换一些数据
将其放入其他MongoDB中
我试图在StreamExecutionEnvironment中使用RuntimeExexutionMode.BATCH运行它，但是应用程序抛出一个异常，因为将我的源代码检测为无界的.而且我不能设置它的限制(它必须完成收集所有文件在mongo集合)
例外情况：
exception in thread "main" java.lang.IllegalStateExcep
浏览 5提问于2021-12-03得票数 0
回答已采纳
我想设置Flink，这样它就可以将数据流从Apache转换并重定向到MongoDB。出于测试目的，我构建在flink流-Connectors.kafka示例()之上。
卡夫卡流是正确红色的链接，我可以映射他们等，但问题发生时，我想保存每一个接收和转换的信息到MongoDB。我发现的关于MongoDB集成的唯一例子是来自github的flink-mongodb测试。不幸的是，它使用的是静态数据源(数据库)，而不是数据流。
我认为应该有一些DataStream.addSink实现的MongoDB，但显然没有。
实现这一目标的最佳途径是什么？我是否需要编写自定义接收器函数，还是可能遗漏了什么？也许应
浏览 3提问于2016-02-02得票数 11
回答已采纳
我正在尝试设置由2个主节点和1个工作节点组成的Apache Flink独立集群。使用Flink 1.6和Zookeeper。要启动和停止cluster，我使用了Flink的1.6文档中描述的过程，即，要启动cluster，我运行start-zookeeper-quorum.sh，然后运行start-cluster.sh，然后运行stop-cluster.sh来停止cluster
在运行一个作业(失败)，然后停止并重新启动集群后，我注意到错误，其中2个作业管理器都无法启动，因为它们正在寻找不存在的目录job_e44fdee88a931200953fed45883ee3f1 (我假设这是我失败作业
浏览 24提问于2018-09-07得票数 0
1回答
Apache通过从检查点恢复作业来保证在故障和恢复时处理一次，检查点是分布式数据流和操作符状态的一致快照(Chandy-Lamport算法用于分布式快照)。此保证在故障转移时恰好是一次。
在正常的集群操作中，Flink如何保证只处理一次，例如给定从外部源读取的Flink源(例如Kafka)，Flink如何保证事件从源读取一次？事件源和Flink源之间是否存在某种类型的应用程序级别？另外，Flink如何保证事件准确地从上游运营商到下游运营商之间只发生一次？对于接收到的事件，这也需要任何形式的附加条件吗？
浏览 3提问于2020-09-30得票数 3
回答已采纳
1回答
我正在对flink中的两个表执行一个左外部联接，下面给出了代码，显示了一个异常，使用相同两个表的内部连接工作良好，能够转换为Datastream。
Table table = customerTable.leftOuterJoin(contactTable,$("cust_custcode")
.isEqual($("contact_custcode")))
.select($("customermessage"), $("contactmessage"));
例外情况是: org.apache.flink.table.api.T
浏览 8提问于2022-08-11得票数 0
1回答
闪存程序的执行必须被触发，例如使用execute()。否则Flink只会创建一个新的执行计划，对吗？我的问题是:在不触发执行的情况下处理延迟操作时，Flink的哪些组件被激活？
根据，有一个优化器负责构建数据流图。是否有更多的过程涉及？
是否有一种方法可以找到优化程序进程的id以便监视它？
浏览 2提问于2015-12-22得票数 1
回答已采纳
2回答
我试图用Flink编写一个计算，它需要两个阶段。
在第一阶段，我从文本文件开始，并执行一些参数估计，从而获得一个表示数据统计模型的Java对象。
在第二阶段，我想使用这个对象来生成模拟数据。
我不知道该怎么做。我尝试使用LocalCollectionOutputFormat，它在本地工作，但是当我在集群上部署作业时，我得到了一个NullPointerException --这并不奇怪。
做这件事的方式是什么？
这是我的代码：
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
GlobalConf
浏览 3提问于2015-04-30得票数 4
回答已采纳
是否可以使用JDBC连接器将Flink Datastream写入Bigquery或任何其他选项？
对于Apache来说，任何建议/示例都是非常有用的。
浏览 18提问于2022-03-14得票数 0
1回答
我正在阅读有关flink的文档--一次功能。我不太明白其中的一些句子：
在成功的预提交之后，提交必须保证最终成功--我们的操作员和外部系统都需要这样的保证。如果提交失败(例如，由于间歇性网络问题)，整个Flink应用程序将失败，根据用户的重新启动策略重新启动，然后再次尝试提交。此过程至关重要，因为如果提交最终不成功，则会发生数据丢失。
这意味着，如果提交最终没有成功，就会发生数据丢失。我将其解释为:提交可能成功，但由于某些原因，每次重新启动都会失败。在这种情况下，Flink只能放弃属于此提交的数据。因此，如果数据丢失是不可接受的，那么应用程序应该重新启动，直到提交成功？
如我们
浏览 2提问于2019-01-30得票数 1
回答已采纳
我想从一个url中获得JSON数组形式的输入流。如何设置源，以便在apache flink中使用datastream.In short连续获得输入我希望从url连续获得json数据，而不关闭flink作业。
浏览 3提问于2018-03-08得票数 1
1回答
我正在运行卡夫卡和Flink作为码头集装箱在我的mac。
我已经实现了Flink作业，它应该使用来自Kafka主题的消息。我运行了一个python生成器，它向主题发送消息。
作业开始时没有问题，但零消息到达。我相信消息被发送到正确的主题，因为我有能够使用消息的python使用者。
flink作业(java):
package com.p81.datapipeline.swg;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataSt
浏览 2提问于2021-11-24得票数 0
回答已采纳
我写了一个模式。我有一个条件列表(来自json的获取规则).Data(Json)来自kafka服务器。我想用这个列表过滤数据。但它不起作用。我该怎么做呢？对于keyedstream和警报，我不是很确定。flink能像这样工作吗？
主程序：
package cep_kafka_eample.cep_kafka;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.gson.Gson;
import c
浏览 1提问于2017-09-21得票数 0
1回答
我使用Flink DataStreams来加入两个流(一个是图书流，一个是出版商流)。我正在尝试使用evictor删除元素，以防它们被从数据库中删除，这由变量deleted表示。 当我在没有evictor的情况下运行代码时，它工作得很好，但是当我添加evictor时，它失败了。 DataStream<BooksWithPublishers> book_publisher = bookStream
.join(publishStream)
.where(value -> value.publisherId)
.equalTo(v
浏览 17提问于2021-04-30得票数 0
回答已采纳
当使用带有buffer-flush选项的upsert-kafka接收器时，我得到下面的错误，而没有buffer-flush选项也是一样。 Exception in thread "main" java.lang.IllegalStateException: There is no the LegacySinkTransformation.
at org.apache.flink.streaming.api.datastream.DataStreamSink.getTransformation(DataStreamSink.java:71)
at org.apache.flin
浏览 72提问于2021-11-08得票数 0
我是flink的新手，我部署了我的flink应用程序，它基本上执行简单的模式匹配。它部署在库伯奈特斯集群，拥有1 JM和6 TM。我每10分钟发送一次4.4k和200k大小的消息，并执行负载测试。我添加了重新启动策略和检查指向，如下所示，我没有显式地使用代码中的任何状态，因为不需要它。
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// start a checkpoint every 1000 ms
env.enableCheckpointing(inter
浏览 4提问于2020-12-04得票数 0
我已经开发了一个Flink程序，可以读取Twitter上的推文，并将它们推送到Kafka上。然后，它从Kafka获取tweet并对其进行处理。
"Tweets“转换从tweet的文本中提取标签和用户，并在默认输出中发出它们，并在副输出中发出它们的每一对。
附加的图像是从Flink Web UI中拾取的。我不明白为什么Kafka源和Tweets处理操作符合并到一个任务中，主要是我希望Tweets接收器接收所有来自Kafka源的原始Tweet，而不是tweets处理操作符的输出。
程序是正确的吗？
这是代码的相关部分：
FlinkKafkaConsumer010<Strin
浏览 1提问于2018-06-30得票数 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
回答已采纳
在Apache中，很容易通过CassandraSink向Cassandra插入一行。但我找不到办法删除一排。
我也试图写自定义接收器，但我得到了NotSerializableException。如何构造删除操作的代码？
public class MyCassandraSink implements SinkFunction<String> {
private Cluster cluster = Cluster.builder()
.addContactPoint("127.0.0.1")
.build();
浏览 5提问于2019-11-09得票数 0
回答已采纳
我计划了一个从Apache读取的应用程序，经过(可能耗时的)处理后，将数据保存到数据库中。
我的例子是消息，而不是流，但为了可伸缩性，我正在考虑将其插入Spark或Flink，但无法理解这些扩展:我的应用程序，当Spark/Flink的一部分读取了Kafka的一些数据，然后退出或继续阅读吗？
那么，Spark/Flink将如何决定他们必须生成更多的我的应用程序实例来提高吞吐量呢？
谢谢!
浏览 4提问于2017-09-20得票数 0
回答已采纳
1回答
我将avro序列化数据发布到kafka主题，然后尝试通过SQL接口从主题创建Flink表。我能够创建主题，但在执行SQL SELECT语句后无法查看主题数据。但是，我能够使用简单的kafka用户反序列化和打印已发布的数据。在SQL上获取此错误：
Flink SQL> SELECT * FROM test_flink2;
[ERROR] Could not execute SQL statement. Reason:
java.lang.ArrayIndexOutOfBoundsException: Index -3 out of bounds for length 2
表创建
Flink
浏览 4提问于2021-02-05得票数 1
回答已采纳
下面的代码片段将演示如何以流模式将数据写入文件。但是，当应用程序开始运行时，我检查了输出目录，发现没有写入数据，_SUCCESS也不存在
每个分区(例如：p_day=2021-04-30\p_hour=18\p_min=52)包含大约58个文件(每个文件是空文件，每秒钟创建一个文件)，名称类似于.part-01c10c8a-5ffa-4ffc-91d0-57ed68d85c93-0-0.inprogress.1cc628a9-0f88-46ca-9d14-7f10db332184。如果我用'format' = 'parquet',更改'format
浏览 0提问于2021-04-30得票数 1
我正在尝试将检查点存储在azure BLOB存储中。根据flink 1.9-SNAPSHOT文档，这应该是可能的。但是，当我尝试添加SBT依赖项时，我得到以下错误： SBT依赖项："org.apache.flink“%% "flink-azure-fs-hadoop”%"1.9-SNAPSHOT“ 错误:找不到警告模块: org.apache.flink#flink-azure-fs-hadoop_2.12；1.9- C:\Users\arshad.ivy2\local\org.apache.flink\flink-azure-fs-hadoop_2.12\1.9-S
浏览 12提问于2019-06-30得票数 0
当我试图在上执行示例时，我会得到以下错误。
通过在的帮助下添加一些额外的参数，我成功地执行了文档中的第一个命令。
user@local:~/flink-1.14.4$ ./bin/kubernetes-session.sh \
-Dkubernetes.cluster-id=dproc-example-flink-cluster-id \
-Dtaskmanager.memory.process.size=4096m \
-Dkubernetes.taskmanager.cpu=2 \
-Dtaskmanager.numberOfTaskSlots=4 \
-Dresourcemanager.
浏览 31提问于2022-03-25得票数 1
我正在尝试实现一个pyflink作业(通过表API)，它在源数据转换为标准格式之后，从多个源进行一些基本处理。我能够将每个源的数据转换为所需的格式(具有指定列的“表模式”)，但不能将数据设置为“处理表”以供进一步实现。这就是我想要的：
要将不同的源模式转换为标准的处理表方案(然后可以加窗)，可以实现一个表函数(用于多个输入行)等等。
编辑：
谢谢你的建议，避免可能保留的名字。更改它，并在udf实现后包括别名(将行结果拆分为列值/名称).已经取得了一些进展，我现在正在讨论的例外是“中央合并表”定义(图表中的标准表)：
py4j.protocol.Py4JJavaError: A
浏览 18提问于2021-12-06得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
