什么是Streaming API
什么是Streaming API
Streaming API是一种实时数据传输的API，它允许开发者通过实时流的方式传输数据，例如音频、视频、实时数据等。Streaming API可以应用于各种场景，例如在线直播、实时音视频通信、实时数据分析等。
Streaming API的优势在于它可以提供低延迟、高质量的实时数据传输，同时也可以降低服务器的负担和成本。它可以帮助开发者快速构建实时数据传输的应用程序，同时也可以提高应用程序的用户体验。
Streaming API的应用场景包括在线直播、实时音视频通信、实时数据分析、物联网设备数据传输、智能家居设备数据传输等。
推荐的腾讯云相关产品包括：
- 腾讯云直播：提供实时音视频传输服务，支持多种分辨率、码率、协议等，可以满足各种直播场景的需求。
- 腾讯云推流：提供实时音视频推流服务，支持多种协议、码率、分辨率等，可以满足各种推流场景的需求。
- 腾讯云实时音视频通信：提供实时音视频通信服务，支持多种音视频编解码器、协议、码率、分辨率等，可以满足各种音视频通信场景的需求。
产品介绍链接地址：
相关·内容
问题
在不同的包中ProcessWindowFunction的区别是什么?在哪里/什么时候使用哪一个包？
org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction
org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
对于窗口操作，就像在org.apache.flink.streaming.api.scala中一样，是正确使用的，但是找不到FLink1.8API文档，只有1.3。
如果我使用org
浏览 1提问于2019-07-07得票数 0
作为标题，异常发生在键控窗口中，
java.lang.IllegalArgumentException: Key Group 91 does not belong to the local range.
at org.apache.flink.util.Preconditions.checkArgument(Preconditions.java:139)
at org.apache.flink.streaming.api.operators.HeapInternalTimerService.getIndexForKeyGroup(HeapInternalTimerService.java:43
浏览 2提问于2018-07-17得票数 1
我正在尝试连接到本地机器上的Kafka (2.1)，并在随Flink (1.7.2)附带的shell中阅读。
我正在做的事情是：
:require flink-connector-kafka_2.11-1.7.1.jar
:require flink-connector-kafka-base_2.11-1.7.1.jar
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.streaming.util.serialization.SimpleStringSc
浏览 0提问于2019-03-11得票数 1
回答已采纳
2回答
我是Apache Spark的新手，我正在尝试运行Spark Streaming + Kafka集成直接方法示例(JavaDirectKafkaWordCount.java)。
我已经下载了所有的库，但是当我尝试运行时，我得到了这个错误
Exception in thread "main" java.lang.NoSuchMethodError: scala.Predef$.ArrowAssoc(Ljava/lang/Object;)Ljava/lang/Object;
at kafka.api.RequestKeys$.<init>(RequestKeys.sca
浏览 2提问于2016-05-04得票数 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
回答已采纳
我正在尝试实现一个具有JDBC postgresql db源连接的流应用程序。作为开始，我尝试了一个基本查询，但是由于强制转换异常，我无法执行它。
以下是例外情况
Caused by: org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException: Could not forward element to next operator
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushTo
浏览 0提问于2019-11-19得票数 0
import com.typesafe.config.ConfigFactory
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.assigners._
import org.
浏览 26提问于2020-06-26得票数 0
回答已采纳
1回答
我正在尝试运行一个简单的程序，它从一个动态流中读取，做一个琐碎的转换，并将结果写入另一个动态流。
在Flink 1.4.0上本地运行(这是目前EMR支持的版本，因此无法升级)。
以下是代码：
def main(args: Array[String]) {
val env = StreamExecutionEnvironment.getExecutionEnvironment
val consumerConfig = new Properties()
consumerConfig.put(AWSConfigConstants.AWS_REGION, "us-east-1&#
浏览 1提问于2018-06-13得票数 0
回答已采纳
我想做一个流媒体作业。
当我尝试使用start-clusted.sh和Flink在本地运行它时，我没有问题。
但是，我目前正在尝试使用Flink on try (部署在Google上)来运行我的工作，当我试图取消它时，取消状态将永远持续，并且TaskManager中仍有一个插槽。
这是我得到的日志：
2016-10-18 16:56:04,053 INFO org.apache.flink.runtime.taskmanager.Task -
Attempting to cancel task Source: pubSubMessageAcknowledgingSource ->
T
浏览 4提问于2016-10-19得票数 2
回答已采纳
我的Flink管道目前使用一个Pojo，其中包含一些列表和映射( Strings)，按照
public class MyPojo {
private List<String> myList = new ArrayList<>();
private OtherPojo otherPojo = new OtherPojo();
// getters + setters...
}
public class OtherPojo {
private Map<String, String> myMap = new HashMap<
浏览 16提问于2020-01-18得票数 3
回答已采纳
当我试图用flatMap操作符写出集合时，我得到了非法的状态异常(仅在高负载下)：缓冲池被破坏了，我在这里做错了什么？当flink抛出缓冲池错误时？
java.lang.RuntimeException: Buffer pool is destroyed.
at org.apache.flink.streaming.runtime.io.RecordWriterOutput.pushToRecordWriter(RecordWriterOutput.java:110)
at org.apache.flink.streaming.runtime.io.RecordWriter
浏览 1提问于2019-04-02得票数 3
回答已采纳
1回答
我试图在Flink的DataStream上应用每个窗口的功能。以下是我的代码
DataStream<Tuple2<String, String>> data = ...
DataStream<Tuple2<String, String>> freqCityChangeTransactions = data
.keyBy(0)
.timeWindow(Time.seconds(5))
.process(new MyProcessWindowFunction());
以下是我实现的MyProcessWindowFunction
浏览 1提问于2018-10-01得票数 2
回答已采纳
我正在尝试运行流作业，它消耗来自Kafka的消息，转换它们，并将其下沉到Cassandra。
当前代码片段失败
val env: StreamExecutionEnvironment = getExecutionEnv("dev")
env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)
.
.
.
.
val source = env.addSource(kafkaConsumer)
.uid("kafkaSource")
浏览 2提问于2020-05-27得票数 1
2回答
我在流的世界里是个新手，在我的第一次尝试中我遇到了一些问题。 更具体地说，我正在尝试使用Flink在滑动窗口中实现计数和groupBy功能。 我在一个普通的DateStream中做过，但是我不能让它在WindowedStream中工作。 你对我该怎么做有什么建议吗？ val parsedStream: DataStream[(String, Response)] = stream
.mapWith(_.decodeOption[Response])
.filter(_.isDefined)
.map { record =>
(
浏览 18提问于2019-05-02得票数 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
2回答
无法看到星火中卡夫卡流的信息
、、、、
我刚刚开始使用Kafka Stream库对Spark进行测试。
我一直在Jupyter Notebook上运行整个设置。我正在尝试从Twitter Streaming获取数据。
Twitter流媒体代码：
import json
import tweepy
from uuid import uuid4
import time
from kafka import KafkaConsumer
from kafka import KafkaProducer
auth = tweepy.OAuthHandler("key", "key")
auth.set_acces
浏览 4提问于2017-09-30得票数 0
java.lang.IllegalStateException: No ExecutorFactory found to execute the application.
at org.apache.flink.core.execution.DefaultExecutorServiceLoader.getExecutorFactory(DefaultExecutorServiceLoader.java:84)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync
浏览 2提问于2020-07-22得票数 7
我已经编写了一个消费者，它读取卡夫卡主题，并使用StreamSink以拼图格式写入数据。但是我得到了以下错误
java.lang.Exception: org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException: Could not forward element to next operator
at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.checkThrowS
浏览 6提问于2020-02-29得票数 0
当StreamingFileSink尝试重命名进程中的文件时，我偶尔会遇到一个问题。请参见下面的异常：
文件系统挂载在Google Cloud Storage buckets上。它在大多数情况下都工作得很好，但突然失败了。
我已经验证了inprogress文件是否存在。任何提示和建议都将非常有用--谢谢！
2019-11-12 01:26:44.908 [flink-akka.actor.default-dispatcher-77] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Job kafka-ingestio
浏览 25提问于2019-11-12得票数 0
我试图在flink服务器上运行一些代码，但无法
下面是我的代码
DataStream<UserInfo> keyedStream = executionEnvironment
.addSource(new UserDataSource());
keyedStream.assignTimestampsAndWatermarks(new MessageWaterEmitter());
tableEnv.registerDataStream("test", keyedStream, "us
浏览 0提问于2020-12-30得票数 0
回答已采纳
当我执行我的Flink应用程序时，它给出了这个NullPointerException：
2017-08-08 13:21:57,690 INFO com.datastax.driver.core.Cluster - New Cassandra host /127.0.0.1:9042 added
2017-08-08 13:22:02,427 INFO org.apache.flink.runtime.taskmanager.Task - TriggerWindow(TumblingEventTimeWindows(30000), List
浏览 5提问于2017-08-08得票数 8
回答已采纳
1回答
Nginx不识别子文件夹
、、、
我正在尝试部署一个Flask应用程序在一个带有nginx和gunicorn的VPS上。 location /app {
root /var/www/mydomain.com/html/flask_app;
proxy_pass http://127.0.0.1:8000/;
proxy_redirect http://$http_host/ https://$http_host/;
proxy_set_header Host $host;
proxy_set_hea
浏览 33提问于2021-04-05得票数 0
1回答
我有一份spark流媒体的工作，代码如下：
val filterActions = userActions.filter(Utils.filterPageType)
val parseAction = filterActions.flatMap(record => ParseOperation.parseMatch(categoryMap, record))
val finalActions = parseAction.filter(record => record.get("invalid") == None)
val userModels = finalA
浏览 0提问于2018-07-01得票数 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
回答已采纳
我实现了一个ProcessFunction，它使用Guava缓存来过滤传入的事件流。代码如下所示： object myJob {
private def updateCache(cacheObject, someValue) = {}
private def getCacheValue(cacheObject, someKey) = {}
override def run(params, executionEnv) = {
val inputStream = executionEnv.stream
val c = CacheBuilder.newBuilder()
v
浏览 132提问于2019-04-02得票数 0
我正在尝试使用tweepy filter选项检索包含诸如挂钟、墙纸、墙贴等词的推文。我知道“space”的意思是AND，而'，‘的意思是OR。
我参考过的官方文档：
在文档示例中，给定的'twitter api,twitter streaming'应该会得到以下tweet。
1. The Twitter API is awesome
2. The twitter streaming service is fast
3. Twitter has a streaming API
有没有办法把这两个操作数结合起来。
例如："twitter api,streaming&
浏览 1提问于2015-07-01得票数 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检查点，当我们使用java基本类型的值状态时，它可以正常运行，但当使用java对象的值状态时，它会抛出异常。它的关键解决方案是什么？任何想法都将受到感谢。 代码如下： var playUVState: ValueState[BigInteger] = _ override def open(parameters: Configuration): Unit = {
playUVState = getRuntimeContext.getState(new ValueStateDescriptor[BigInteger]("playUVState"
浏览 16提问于2019-05-29得票数 2
1回答
当我在windowfunction之前使用countwindow时，在运行代码后我得到了类似State migration is currently not supported的错误，详细消息是
org.apache.flink.util.FlinkRuntimeException: org.apache.flink.util.StateMigrationException: State migration is currently not supported.
at org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.c
浏览 17提问于2020-02-11得票数 0
嗨，我用下面的代码发布从flink到Kafka的消息。
final Properties props = new Properties();
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALU
浏览 14提问于2022-06-14得票数 0
回答已采纳
我正在尝试使用Scala在我的Apache Flink项目中使用ProcessWindowFunction。不幸的是，我在实现Apache Flink文档中使用的基本ProcessWindowFunction时已经失败了。
这是我的代码：
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
import org.apache.flink.streaming.api.windowing.time.Time
浏览 1提问于2018-11-30得票数 6
我编写了一个代码来使用来自Kafka的流，然后将其接收到MySQL。
代码在我的IDE上运行正常，可以按预期向目标表中插入数据。
但是，当我将jar提交给Flink(部署在192.168.95.2上)网页时，它在preparedStatement = connection.prepareStatement(sql);语句上抛出了NullPointerException。这些部分的其他代码是：
private static PreparedStatement preparedStatement;
sql = "insert into kafka_ccu values(?,?)";
浏览 22提问于2020-03-04得票数 0
我想弄清楚如何写我自己的WindowFunction，但有问题，但我不知道为什么。我遇到的问题是应用函数，因为它不承认MyWindowFunction是一个有效的输入，所以我无法编译。我正在流的数据包含(timestamp,x,y)，其中x和y为0和1用于测试。extractTupleWithoutTs只是返回一个元组(x,y)。我一直在用简单和运行代码，并成功地减少了函数。感谢您的帮助:)使用Flink 1.3
进口：
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.stream
浏览 0提问于2017-11-07得票数 1
回答已采纳
关于，我得到的提示是应该使用getOrCreate惯用法来避免这个问题。但是正在尝试：
JavaStreamingContextFactory contextFactory = new JavaStreamingContextFactory() {
@Override
public JavaStreamingContext create() {
final SparkConf conf = new SparkConf().setAppName(NAME);
return new JavaStreamingContext(conf, Durati
浏览 5提问于2017-01-18得票数 0
1回答
我试图通过提供Kryo序列化程序来解决以下问题，但它仍然不起作用。它无法识别ModelCom的序列化程序。此外，任何打印功能的消息都不会显示出来。 我使用的是Apache Flink 1.9.0和Apache Jena 3.10.0 我的Kotlin代码： val serializer = object : Serializer<Model>(){
override fun write(kryo: Kryo, output: Output?, obj : Model?) {
print("write")
浏览 21提问于2020-04-27得票数 1
回答已采纳
我试着用Spark Streaming从Twitter上流式传输数据。但是
下面的问题。
import org.apache.spark.streaming.twitter._
import twitter4j.auth._
import twitter4j.conf._
import org.apache.spark.streaming.{Seconds,StreamingContext}
import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.Stream
浏览 4提问于2017-06-15得票数 0
错误时:使用Scala和Twitter4j在星火中创建一个tweet流。下面是我的代码片段：
scala> val ssc = new StreamingContext(sc, Seconds(10))
scala> val cb = new ConfigurationBuilder
scala>cb.setDebugEnabled(true).setOAuthConsumerKey("**********").setOAuthConsume
scala> val auth = new OAuthAuthorization(cb.build)
scala
浏览 1提问于2016-04-14得票数 1
回答已采纳
需要帮助！谁能给我指引一条正确的路？
以下是我的代码和日志的片段。
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
1回答
我有一个在生产环境中运行的Flink流作业，我需要对主转换代码进行更改。
生产中的代码实际上如下所示：
stream
.filter(inboundData -> inboundData.hasToBeFiltered())
.uid("filtered-data")
.keyBy(data -> data.getMyStringKey())
.process(doSomething())
.uid("processed-inbound-data-id");
我需要改变keyBy()操作符使用inboundData PO
浏览 0提问于2019-08-08得票数 1
下面是Flink关于如何将WatermarkStrategy与KafkaConsumer结合使用的文档。代码如下所示
val kafkaSource = new FlinkKafkaConsumer[MyType]("myTopic", schema, props)
kafkaSource.assignTimestampsAndWatermarks(
WatermarkStrategy
.forBoundedOutOfOrderness(Duration.ofSeconds(20)))
val stream: DataStream[MyType] = env.ad
浏览 5提问于2020-11-23得票数 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
我是新的火花，我试图创建简单的JavaDStream，以测试我的工作火花测试基API。我迄今所做的是：
JavaStreamingContext streamingContext = new
JavaStreamingContext(jsc(),Durations.seconds(10));
List<String> list = new LinkedList<String>();
list.add("first");
list.add("second");
list.add(
浏览 2提问于2016-06-23得票数 5
回答已采纳
我最近从1.9.0更新到flink 1.10.0，并在尝试本地执行作业时开始出现此错误。令人惊讶的是，它在IDE中工作得很好。只有当我尝试从命令行(java -jar)运行可执行jar时，我才得到这个错误。
，它说要添加一个依赖项，但我已经有了。有什么想法吗？
作为参考，我有：
"org.apache.flink" %% "flink-scala" "1.10.0",
"org.apache.flink" %% "flink-streaming-scala" % "1.10.0",
"org
浏览 0提问于2020-06-05得票数 0
我开始在Java的ApacheFlink中工作。
我的目标是在一分钟的时间窗口中使用一个ApacheKafka主题，这将应用非常基本的信息，并将每个窗口的结果记录在一个文件中。
到目前为止，我成功地将文本转换简化应用于我接收的内容，我应该使用apply或process来写入文件，但窗口的结果我有点迷失。
到目前为止，这是我的代码
package myflink;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunctio
浏览 17提问于2019-11-08得票数 0
1回答
我们在Kubernetes中运行Apache Flink 1.9。我们有一些工作，消费卡夫卡事件，并收集每分钟的计数。这些作业一直运行得很好，但最近突然出现了许多错误 java.lang.RuntimeException: Partition already finished.
at org.apache.flink.streaming.runtime.io.RecordWriterOutput.pushToRecordWriter(RecordWriterOutput.java:110)
at org.apache.flink.streaming.runtime.io.Re
浏览 85提问于2020-10-14得票数 1
回答已采纳
我正在尝试使用KeyedProcessFunction，但是ctx: Context变量在processFunction中的KeyedProcessFunction中返回null。请注意，我使用的是默认的TimeCharacteristic，即ProcessingTime (所以我甚至不设置它)。
我在堆栈溢出上找到了，但它与EventTime有关，而不是ProcessingTime。
按照的确切示例，我使用Scala2.11.12和Flink 1.10创建了以下内容，并且仍然得到相同的错误。
import org.apache.flink.streaming.api.scala._
impo
浏览 2提问于2020-03-18得票数 1
回答已采纳
1回答
我有一个Flink应用程序，它保持与这个错误的错误。 com.org.ads.audience.traffic.MyClass@6eaa21d8 is not serializable. The object probably contains or references non serializable fields.
org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:140)
org.apache.flink.api.java.ClosureCleaner.clean(Cl
浏览 448提问于2019-09-25得票数 1
1回答
我正在尝试使用the documentation of JSForce中提供的代码从Node.js服务器订阅SalesForce主题 conn.streaming.topic("InvoiceStatementUpdates").subscribe(function(message) {
console.log('Event Type : ' + message.event.type);
console.log('Event Created : ' + message.event.createdDate);
console.log(
浏览 9提问于2019-03-18得票数 1
回答已采纳
1回答
我有一份Flink的工作，把数据写进卡夫卡。Kafka主题的最大消息大小设置为5MB，所以如果我尝试编写任何大于5MB的记录，它会抛出以下异常并将作业降低。
java.lang.Exception: Failed to send data to Kafka: The request included a message larger than the max message size the server will accept.
at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducerBase.checkErroneo
浏览 1提问于2018-09-13得票数 1
回答已采纳
我们正在使用包括map、windowing、flatMap()在内的几个运算符运行一个Flink作业，作业失败并出现以下错误-只是想知道是什么导致了这个错误：
2021-05-27 07:00:07,023 WARN org.apache.flink.runtime.taskmanager.Task [] - Collect Inventory (1/1)#0 (34d81cf2e59f350886f93a1e0f734d38) switched from RUNNING to FAILED with failure cause: org.apache.
浏览 229提问于2021-05-27得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
