Flink撤回机制问题之撤回机制不起作用如何解决

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: Apache Flink是由Apache软件基金会开发的开源流处理框架,其核心是用Java和Scala编写的分布式流数据流引擎。本合集提供有关Apache Flink相关技术、使用技巧和最佳实践的资源。

问题一:flink 1.13编译 flink-parquet报错

查看发现 org.apache.avro avro-maven-plugin ${avro.version} generate-sources schema

${project.basedir}/src/test/resources/avro

${project.basedir}/target/generated-test-sources/

这个编译插件的问题 同 flink 1.12 (编译没问题)比较发现: 区别: ${project.basedir}/src/test/java/

当我把flink 1.13/ master 改成这个就可以编译通过了。 我想知道有人遇到跟我一样的问题吗。是我环境的问题吗?*来自志愿者整理的flink邮件归档


参考答案:

是不是没有删除之前生成的类,手动删除冲突的类试试。来自志愿者整理的flink邮件归档



关于本问题的更多回答可点击进行查看:

https://developer.aliyun.com/ask/370031?spm=a2c6h.13066369.question.1.33bf585ffd7RGn



问题二:Dynamic Table Options 被优化器去掉了

我有在使用 temporal Joini 的时候有设置 如果读取分区的相关的 dynamic option,但是最后是没有生效的,我看全部使用的默认参数,打印出来了执行计划,逻辑执行计划是有的,优化之后没有了 如下,我设置的是加载最新分区,24小时加载一次,我看最后运行的日志是加载的全部分区,1小时有一次加载,这都是默认的参数,所以怀疑是 dyanmic option 没有生效。

== Abstract Syntax Tree == +- LogicalSnapshot(period=[$cor0.proctime]) +- LogicalTableScan(table=[[daaaas, my_db, store_da_table, source: [HiveTableSource(store_id, store_name, merchant_id, tag_id, brand_id, tob_user_id, is_use_wallet, is_use_merchant_app, longitude, latitude, state, city, district, address, postal_code, register_phone, email, email_source, register_time, logo, banner, partner_type, commission_rate, tax_rate, service_fee, min_spend, delivery_distance, preparation_time, contact_phone, store_status, closed_start_time, closed_end_time, effective_closed_end_time, auto_confirmed, auto_confirmed_enabled, create_time, update_time, rating_total, rating_score, opening_status, surcharge_intervals, service_charge_fee_rate, driver_modify_order_enabled, delivery_distance_mode, business_info_added, mtime, dt, grass_region) TablePath: my_db.store_da_table, PartitionPruned: false, PartitionNums: null], dynamic options: {streaming-source.enable=true, streaming-source.monitor-interval=24 h, streaming-source.partition.include=latest}]])

== Optimized Logical Plan == Calc(select=[_UTF-16LE'v4' AS version, _UTF-16LE'ID' AS country, city, id, event_time, operation, platform, payment_method, gmv, 0.0:DECIMAL(2, 1) AS gmv_usd], where=[NOT(LIKE(UPPER(store_name), _UTF-16LE'%[TEST]%'))]) +- LookupJoin(table=[daaaas.my_db.store_da_table], joinType=[LeftOuterJoin], async=[false], lookup=[store_id=store_id], select=[city, id, event_time, operation, platform, payment_method, gmv, store_id, store_id, store_name]) +- Union(all=[true], union=[city, id, event_time, operation, platform, payment_method, gmv, store_id]) :- Calc(select=[delivery_city AS city, id, /(CAST(create_time), 1000) AS event_time, CASE(OR(=(order_status, 440), =(order_status, 800)), _UTF-16LE'NET':VARCHAR(5) CHARACTER SET "UTF-16LE", _UTF-16LE'GROSS':VARCHAR(5) CHARACTER SET "UTF-16LE") AS operation, _UTF-16LE'xxxx' AS platform, payment_method, /(CAST(total_amount), 100000) AS gmv, CAST(store_id) AS store_id]) : +- DataStreamScan(table=[[daaaas, keystats, main_db__transaction_tab]], fields=[id, delivery_city, store_id, create_time, payment_time, order_status, payment_method, total_amount, proctime], reuse_id=[1]) +- Calc(select=[delivery_city AS city, id, /(CAST(payment_time), 1000) AS event_time, _UTF-16LE'NET':VARCHAR(5) CHARACTER SET "UTF-16LE" AS operation, _UTF-16LE'AIRPAY' AS platform, payment_method, /(CAST(total_amount), 100000) AS gmv, CAST(store_id) AS store_id], where=[OR(=(order_status, 440), =(order_status, 800))]) +- Reused(reference_id=[1])*来自志愿者整理的flink邮件归档



参考答案:

看样子是个bug,能提供以下你的ddl以及相关的环境吗?方便我们复现下问题。来自志愿者整理的flink邮件归档



关于本问题的更多回答可点击进行查看:

https://developer.aliyun.com/ask/370033?spm=a2c6h.13066369.question.2.33bf585fJQ9OwU



问题三:flink如flink如何从流中取出自定义的数据结构出问题

flink版本使用1.12.2。有一个需求就是想要从stream中拿出自定义的数据结构,暂且叫a并赋值给后面变量,基于这个a取他的属性作一些判断操作。 比如: val ds: DataStream[b] = stream.filter(_.nonEmpty).map(new MapFunction[String, b] {

override def map(value: String) = { val recallKafka = JSON.parseObject(value, classOf[a])

b(recallKafka.group_id, value, recallKafka.eventTime)

} })

val kafkaCommonData: a =recallKafka 判断条件 if (kafkaCommonData.data.date != null) {xxxxx} if (kafkaCommonData.data.userinfo != null) {xxxx} ..... 请问一下,我通过什么方法能单独把流中的某个数据结构给取出来呢?如果有方式的话应该要怎么写呢?大佬们帮忙看一下啊,卡了好几天 了,难受。*来自志愿者整理的flink邮件归档



参考答案:

你可以用 filter 过滤出多个流或者用测流输出的方式分流处理。来自志愿者整理的flink邮件归档



关于本问题的更多回答可点击进行查看:

https://developer.aliyun.com/ask/370037?spm=a2c6h.13066369.question.3.33bf585fZX2f7T



问题四:Flink撤回机制不起作用

hi,all 我想简单的测试一下撤回机制,于是我写了以下代码 tableEnv.createTemporaryView("sensor", sensorTable); val resultSqlTable = tableEnv.sqlQuery("select country, count(order_id) as cnt from sensor group by country"); tableEnv.toRetractStreamWaterSensorCnt.print("result");

然后在socket发送以下数据: 001 usa 002 usa 003 china 002 china 004 usa

我预期在控制台得到的结果应该是 usa, 2 china, 2

但是结果却是: usa, 3 china, 2

本应该usa撤回一条才符合我对撤回机制的理解,但是usa并没有减少? 大家可以帮我消除疑惑吗? 如果您能在百忙之中抽空解答,我将非常感激!*来自志愿者整理的flink邮件归档



参考答案:

这个应该是上次计算的结果保留下来,而下一次并不会对原先的数据进行重新计算的。来自志愿者整理的flink邮件归档



关于本问题的更多回答可点击进行查看:

https://developer.aliyun.com/ask/370038?spm=a2c6h.13066369.question.6.33bf585fuhiGtq



问题五:官网文档和样例的不完整性和不严谨性的问题

Flink1.10的集群,用hdfs做backend

无论从flink最早的版本到flink 1.12都存在的一些文档和样例的不完整,或者说相同的代码,因输入源不同导致的结果差异。

比如说下面链接中的样例 https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/stream/operators/process_function.html

如果输入源分别为

  1. 一次性从内存中的List读取数据
  2. 一次性从文件目录读取读取数据
  3. 持续从文件目录读取数据
  4. 从socket流持续读取文件

上面的4者,只有3和4,对于KeyedStream的process(…)中使用ValueState 在处理onTimer函数时才会被触发调用,对于1和2是不会的。

相信其他的算子也存在类似的问题

具体代码如下:

package com.xxx.data.stream;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.io.TextInputFormat;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
import org.apache.flink.streaming.api.functions.IngestionTimeExtractor;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.util.Collector;
import javax.annotation.Nullable;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
public class KeyedStreamJob {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
env.setParallelism(4);
//1.从内存获取数据
Tuple2<String, Integer> item = null;
List<Tuple2<String, Integer>> items = new ArrayList<>();
item = new Tuple2<>("k1", 1);
items.add(item);
item = new Tuple2<>("k3", 3);
items.add(item);
item = new Tuple2<>("k1", 10);
items.add(item);
item = new Tuple2<>("k2", 2);
items.add(item);
item = new Tuple2<>("k1", 100);
items.add(item);
item = new Tuple2<>("k2", 20);
items.add(item);
DataStreamSource<Tuple2<String, Integer>> streamSource = env.fromCollection(items);
SingleOutputStreamOperator<Tuple2<String, Integer>> listStream = streamSource.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Integer>>() {
@Nullable
@Override
public Watermark getCurrentWatermark() {
return null;
}
@Override
public long extractTimestamp(Tuple2<String, Integer> element, long previousElementTimestamp) {
System.out.println("---");
return System.currentTimeMillis();
}
});
//2.从文件夹一次性获取数据
SingleOutputStreamOperator<Tuple2<String, Integer>> fileStream = env.readTextFile("D:\\data", "UTF-8").map(new MapFunction<String, Tuple2<String, Integer>>() {
@Override
public Tuple2<String, Integer> map(String value) throws Exception {
return new Tuple2<>(value, 1);
}
})
.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Integer>>() {
@Nullable
@Override
public Watermark getCurrentWatermark() {
return null;
}
@Override
public long extractTimestamp(Tuple2<String, Integer> element, long previousElementTimestamp) {
return System.currentTimeMillis();
}
});
//3.从文件夹持续获取数据
TypeInformation<String> typeInformation = BasicTypeInfo.STRING_TYPE_INFO;
TextInputFormat format = new TextInputFormat(new Path("D:\\data"));
format.setCharsetName("UTF-8");
//是否支持递归
format.setNestedFileEnumeration(true);
SingleOutputStreamOperator<Tuple2<String, Integer>> continuefileStream = env.readFile(format, "D:\\data", FileProcessingMode.PROCESS_CONTINUOUSLY, 6000L, typeInformation).map(new MapFunction<String, Tuple2<String, Integer>>() {
@Override
public Tuple2<String, Integer> map(String value) throws Exception {
return new Tuple2<>(value, 1);
}
})
.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Integer>>() {
@Nullable
@Override
public Watermark getCurrentWatermark() {
return null;
}
@Override
public long extractTimestamp(Tuple2<String, Integer> element, long previousElementTimestamp) {
return System.currentTimeMillis();
}
});
//4.从socket中持续获取数据
SingleOutputStreamOperator<Tuple2<String, Integer>> socketStream = env.socketTextStream("localhost", 9999).map(new MapFunction<String, Tuple2<String, Integer>>() {
@Override
public Tuple2<String, Integer> map(String value) throws Exception {
return new Tuple2<>(value, 1);
}
})
.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Integer>>() {
@Nullable
@Override
public Watermark getCurrentWatermark() {
return null;
}
@Override
public long extractTimestamp(Tuple2<String, Integer> element, long previousElementTimestamp) {
return System.currentTimeMillis();
}
});
//分别从1. 2. 3. 4. 测试数据的ValueState的超时触发,发现
//只有3.continuefileStream 4.socketStream 这些持续获取数据的可以触发onTimer
//至于1.listStream 2.fileStream 这些一次性获取书的不会触发onTimer
SingleOutputStreamOperator<Tuple2<String, Long>> sum = continuefileStream // listStream fileStream socketStream
.keyBy(0)
.process(new KeyedProcessFunction<Tuple, Tuple2<String, Integer>, Tuple2<String, Long>>() {
private ValueState<SumWithTimeStamp> sum;
private final SimpleDateFormat yyyyMMddHHmmss = new SimpleDateFormat("yyyy-MM-dd:HH-mm-ss.SSS");
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
StateTtlConfig stateTtlConfig = StateTtlConfig.newBuilder(Time.seconds(1L)).returnExpiredIfNotCleanedUp().updateTtlOnReadAndWrite().useProcessingTime().build();
ValueStateDescriptor<SumWithTimeStamp> valueStateDescriptor = new ValueStateDescriptor<SumWithTimeStamp>("sum", SumWithTimeStamp.class);
// valueStateDescriptor.enableTimeToLive(stateTtlConfig);
sum = getRuntimeContext().getState(valueStateDescriptor);
}
@Override
public void processElement(Tuple2<String, Integer> item, Context ctx, Collector<Tuple2<String, Long>> out) throws Exception {
SumWithTimeStamp sumValue = sum.value();
if (sumValue == null) {
sumValue = new SumWithTimeStamp();
sumValue.key = item.f0;
// Thread.sleep(1500L);
// Date cur = new Date();
// cur.setTime(ctx.timestamp());
// System.out.println("ini " + ctx.getCurrentKey().toString() + yyyyMMddHHmmss.format(cur));
sumValue.sum += item.f1.longValue();
sumValue.lastModified = ctx.timestamp();
sum.update(sumValue);
ctx.timerService().registerProcessingTimeTimer(sumValue.lastModified + 3*1000);
System.out.println("ini " + ctx.getCurrentKey().toString() + " item:" + item.toString() + " sum:" + sum.value().sum);
} else {
sumValue.sum += item.f1.longValue();
sumValue.lastModified = ctx.timestamp();
sum.update(sumValue);
// ctx.timerService().registerProcessingTimeTimer(sumValue.lastModified + 5*1000);
System.out.println("up " + ctx.getCurrentKey().toString() + " item:" + item.toString() + " sum:" + sum.value().sum);
}
// Date cur = new Date();
// cur.setTime(ctx.timestamp());
// System.out.println("up " + ctx.getCurrentKey().toString() + yyyyMMddHHmmss.format(cur));
// Thread.sleep(1500L);
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, Long>> out) throws Exception {
// super.onTimer(timestamp, ctx, out);
System.out.println("-------" + ctx.getCurrentKey().toString());
if (timestamp <= sum.value().lastModified + 5000) {
out.collect(new Tuple2<String, Long>(sum.value().key, sum.value().sum));
// sum.clear();
}
}
});
sum.print();
//continueSum(streamSource);
env.execute("keyedSteamJob");
// System.in.read();
}
public static void continueSum(DataStreamSource<Tuple2<String, Integer>> streamSource) {
streamSource
//by 1
//.assignTimestampsAndWatermarks(new IngestionTimeExtractor())
.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
@Override
public String getKey(Tuple2<String, Integer> value) throws Exception {
return value.f0;
}
})
// .window(TumblingEventTimeWindows.of(Time.milliseconds(10L)))
.sum(1)
.print("+++++++++++++++++++++++++++");
}
public static class SumWithTimeStamp {
public String key;
public long sum;
public long lastModified;
}
}*来自志愿者整理的flink邮件归档



参考答案:

非常感谢你指出文档的问题!

由于邮件中看代码比较吃力(没有语法高亮以及排版的问题),我只是粗略地看了下代码。

当输入源 为 一次性从内存中的List读取数据,无法触发onTimer。 实际的例子中,我看到看到采用的是process time,且延时 3s 触发 。我怀疑是不是,数据量太少,所以程序很快就结束了导致没来得及触发timer,建议改成event time试试这种情况。来自志愿者整理的flink邮件归档



关于本问题的更多回答可点击进行查看:

https://developer.aliyun.com/ask/370039?spm=a2c6h.13066369.question.7.33bf585f7bGBBy

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
相关文章
|
SQL 双11 流计算
Flink SQL 功能解密系列 —— 流计算“撤回(Retraction)”案例分析
通俗讲retract就是传统数据里面的更新操作,也就是说retract是流式计算场景下对数据更新的处理方式。
|
6月前
|
SQL Oracle 关系型数据库
实时计算 Flink版产品使用合集之可以按日期统计数据并且能够撤回已落库数据吗
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
2月前
|
运维 数据处理 数据安全/隐私保护
阿里云实时计算Flink版测评报告
该测评报告详细介绍了阿里云实时计算Flink版在用户行为分析与标签画像中的应用实践,展示了其毫秒级的数据处理能力和高效的开发流程。报告还全面评测了该服务在稳定性、性能、开发运维及安全性方面的卓越表现,并对比自建Flink集群的优势。最后,报告评估了其成本效益,强调了其灵活扩展性和高投资回报率,适合各类实时数据处理需求。
|
4月前
|
存储 监控 大数据
阿里云实时计算Flink在多行业的应用和实践
本文整理自 Flink Forward Asia 2023 中闭门会的分享。主要分享实时计算在各行业的应用实践,对回归实时计算的重点场景进行介绍以及企业如何使用实时计算技术,并且提供一些在技术架构上的参考建议。
813 7
阿里云实时计算Flink在多行业的应用和实践
|
6天前
|
存储 分布式计算 流计算
实时计算 Flash – 兼容 Flink 的新一代向量化流计算引擎
本文介绍了阿里云开源大数据团队在实时计算领域的最新成果——向量化流计算引擎Flash。文章主要内容包括:Apache Flink 成为业界流计算标准、Flash 核心技术解读、性能测试数据以及在阿里巴巴集团的落地效果。Flash 是一款完全兼容 Apache Flink 的新一代流计算引擎,通过向量化技术和 C++ 实现,大幅提升了性能和成本效益。
538 10
实时计算 Flash – 兼容 Flink 的新一代向量化流计算引擎
|
3月前
|
SQL 消息中间件 Kafka
实时计算 Flink版产品使用问题之如何在EMR-Flink的Flink SOL中针对source表单独设置并行度
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
3天前
|
SQL 运维 数据可视化
阿里云实时计算Flink版产品体验测评
阿里云实时计算Flink基于Apache Flink构建,提供一站式实时大数据分析平台,支持端到端亚秒级实时数据分析,适用于实时大屏、实时报表、实时ETL和风控监测等场景,具备高性价比、开发效率、运维管理和企业安全等优势。
|
19天前
|
运维 搜索推荐 数据安全/隐私保护
阿里云实时计算Flink版测评报告
阿里云实时计算Flink版在用户行为分析与标签画像场景中表现出色,通过实时处理电商平台用户行为数据,生成用户兴趣偏好和标签,提升推荐系统效率。该服务具备高稳定性、低延迟、高吞吐量,支持按需计费,显著降低运维成本,提高开发效率。
51 1
|
22天前
|
运维 数据处理 Apache
数据实时计算产品对比测评报告:阿里云实时计算Flink版
数据实时计算产品对比测评报告:阿里云实时计算Flink版
|
2月前
|
存储 运维 监控
阿里云实时计算Flink版的评测
阿里云实时计算Flink版的评测
67 15

相关产品

  • 实时计算 Flink版