问题一: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
如果输入源分别为
- 一次性从内存中的List读取数据
- 一次性从文件目录读取读取数据
- 持续从文件目录读取数据
- 从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