回复： CheckPoint Dir 路径下引发的一些问题

Hi 麻烦使用第三方图床，把图片链接过来，直接贴图片的话显示不出来 --原始邮件-- 发件人:"Weihua Hu"

Re: CheckPoint Dir 路径下引发的一些问题

HI, Px New 1. Checkpoint 保存数量可以通过参数： state.checkpoints.num-retained 来控制，默认是 1 2. _metadata 只是一些元数据，保存了state 的句柄，其他文件是 state 数据，由各 Task 在触发 checkpoint 的时候上传。相反在恢复 checkpoint 的时候JM 读取_metadata 将相应句柄下发到 Task，Task 通过远端 HDFS 拉取对应的 state。 Best Weihua Hu > 2020年6月5日 13:36，Px New

Re: flink sql ddl能设置ttl吗？

Hi, 第一个问题，即将发布的1.11可以设定primary key, 不用再推断primary key，除pk外的字段也没有限制，当前你可以加firt_value,last_value等函数取非group by的其他字段, jdbc的文档在撰写中了[1] 第二个问题，目前应该还不支持通过SET在DDL里设置ttl的. Best, Leonard Xu [1] https://issues.apache.org/jira/browse/FLINK-17829 [2]

CheckPoint Dir 路径下引发的一些问题

Hi everyOne 有一个关于CheckPoint相关的一个问题: 1.我在项目中使用的状态后端为:Fsstatebackend 2.我在jobManager的log输出找到相应的job ID后 去对应的HDFS 找到了对应的chk目录 3.但我有两个疑问: 3.1.没有设置 chk的存储数默认是多保留多少份呢(我这边看到保留了近20次的chk)? 3.2 当我点进具体的chk-id 后 发现有很多文件[见2图] 我清楚的是当任务发生异常后tesk 会从hdfs 将_metadata 下载后进行任务恢复操作的,那其他的哪些文件是如何产生的?以及有什么作用呢? 期待回复:

Re: flink sql ddl能设置ttl吗？

Hi， 1. 如果我没记错的话，mysql的写入应该就是replace的方式写入的，应该不需要group by来实现upsert写入吧，除非你想要的是聚合结果。 2. 这个看你用的是哪种方式，如果是table api，可以直接用你说的这种方式来设置；如果用的是sql-client，可以直接在配置文件里面配置。 xu yihan 于2020年6月4日周四 下午6:07写道： > 各位好， > > 我有一个需求是要维表关联mysql数据，再upsert写入mysql，就是个丰富字段的简单需求。 > > 现在有两个问题： >

Re: Flink sql nested elements

Hi，Ramana For nested data type, Flink use dot (eg a.b.c) to visit nested elements. Your SQL syntax looks right, which Flink version are you using? And could you post your Avro Schema file and DDL ? Best, Leonard Xu > 在 2020年6月5日，03:34，Ramana Uppala 写道： > > We have Avro schema that contains

Re: Batch 模式 Table API 增加cache算子

这个内容会放在 FLIP-36 [1]中讨论，估计会在 1.12 的计划中。 cc @Becket Qin who is the driver of this FLIP. Best, Jark [1] FLIP-36: https://cwiki.apache.org/confluence/display/FLINK/FLIP-36%3A+Support+Interactive+Programming+in+Flink [2] 讨论邮件:

Re: flink1.10整合hbase测试遇到的问题

FYI Leonard's reply in another thread: = Flink hbase connector官网有的[1],1.10以前的包名是flink-hbase, 1.11后社区统一了所有connector的包名，包名为flink-connector-hbase[1]，1.11最近快发布了，你的报错看起来是缺少了一些类，可以参考下hbase connector的依赖[2]。 [1]

Re: SQL Expression to Flink FilterFunction?

This is possible but may need some development. There is a similar util in table tests called `org.apache.flink.table.expressions.utils.ExpressionTestBase` [1], it converts/translates expressions (either Table API Expression or SQL expression) into a MapFunction. I think you can imitate the way

Re: Multiple Sinks for a Single Soure

Hi Prasanna, That’s good to hear and thanks for confirming that it works :) Piotrek > On 3 Jun 2020, at 16:09, Prasanna kumar wrote: > > Piotr and Alexander , > > I have fixed the programmatic error in filter method and it is working now. > > Thanks for the detailed help from both of you.

Re: Re: standalone模式下metaspace内存溢出

你好， 社区已经发现过一些与 metaspace 泄漏相关的问题，其中部分已经修复，还有一些则与第三方依赖有关。由于你的图片没有显示出来，我不确定你遇到的是否是相同的问题。 此外，在即将发布的 1.11 中，Flink 针对这一问题进行了优化，作业会采用单独的 ClassLoader 运行，以避免作业结束之后 metaspace 还有泄漏的问题。1.11.0 已经进入发布前的测试阶段，RC1 已经发布，欢迎试用。 Thank you~ Xintong Song On Fri, Jun 5, 2020 at 9:29 AM 胡泽康 wrote: > flink版本：1.10

Re: Flink 1.6 版本，RocksDBStateBackend ,增量 Checkpoint 目录问题

对 我认为这也不是很合理, 不知道有什么更好的解释吗 LakeShen 于2020年1月16日周四 下午5:06写道： > 原来是这样，中间的确有很多 Checkpoint 超时失败，对于这种由于 Checkpoint 失败创建的目录，Flink 本身是不会删除的，对吗？ > 也就是这部分状态文件，会一直存储在HDFS 上面，这种情况是否会造成 Flink Checkpoint 目录下 shared目录无限增大呢， > 非常感谢你，lucas，期待你的回复。 > > lucas.wu 于2020年1月16日周四 下午4:50写道： > > >

??????flink1.9 Sql ????????????????????????state??????

?? ---- ??:"zhiyezou"<1530130...@qq.com; :2020??6??5??(??) 10:31 ??:"user-zh"

??????flink1.9 Sql ????????????????????????state??????

Hi ??TTL ---- ??:"star"<3149768...@qq.com; :2020??6??5??(??) 10:22 ??:"user-zh@flink.apache.org"

Re: standalone模式下metaspace内存溢出

那我先试试1.11. thank you~ --Original-- From: "Xintong Song"

flink1.9 Sql ????????????????????????state??????

?? flink 1.9??blink planner ?? ?? id hbasemonthtable rowkey??month+city ?? city ??hbasetotalTable ??rowkey?? city 18?? restore??ck??

Re: Flink Dashboard UI Tasks hard limit

Hi Vijay, >From the information you provided (the configurations, error message & screenshot), I'm not able to find out what is the problem and how to resolve it. The error message comes from a healthy task manager, who discovered that another task manager is not responding. We would need to

Re: SQL Expression to Flink FilterFunction?

Hi, Theo Currently, It’s hard to do this in your DataStream application from my understanding, because converting sql expression to Flink operator happens in underlying table planner (more precisely in code generate phase) and it does not expose interface to user so that you can not assign

Re: standalone模式下metaspace内存溢出

你好， 社区已经发现过一些与 metaspace 泄漏相关的问题，其中部分已经修复，还有一些则与第三方依赖有关。由于你的图片没有显示出来，我不确定你遇到的是否是相同的问题。 此外，在即将发布的 1.11 中，Flink 针对这一问题进行了优化，作业会采用单独的 ClassLoader 运行，以避免作业结束之后 metaspace 还有泄漏的问题。1.11.0 已经进入发布前的测试阶段，RC1 已经发布，欢迎试用。 Thank you~ Xintong Song On Fri, Jun 5, 2020 at 9:29 AM 胡泽康 wrote: > flink版本：1.10

Re: Native K8S not creating TMs

If you have created the role binding "flink-role-binding-default" successfully, then it should not be the RBAC issue. It seems that kubernetes-client in JobManager pod could not contact to K8s apiserver due to okhttp issue with java 8u252. Could you add the following config option to disable

回复： flink sql upsert模式写入mysql，es等key一定是groupby之后所有字段吗

Hi 这里存在一个问题是，使用了last_value或者first_value这样的udaf，但是如果多条数据来到经过udaf处理后结果还是和之前一样的情况下，是不会产出回撤流数据的，可以观察下你是否需要考虑这种情况。 Best, Yichao Yang --原始邮件--

Re: flink整合hbase

Hi, Flink hbase connector官网有的[1],1.10以前的包名是flink-hbase, 1.11后社区统一了所有connector的包名，包名为flink-connector-hbase[1]，1.11最近快发布了，你的报错看起来是缺少了一些类，可以参考下hbase connector的依赖[2]。 Best, Leonard Xu [1] https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connect.html#hbase-connector

Re: FlinkKinesisProducer sends data to only 1 shard. Is it because I don't have "AggregationEnabled" set to false ?

Hi, Looks like I am sending a Map to Kinesis and it is being sent to 1 partition only. *How can I make this distribute across multiple partitions/shards on the Kinesis Data stream with this Map* data ? *Sending to Kinesis*: DataStream> influxToMapKinesisStream = enrichedMGStream.map(influxDBPoint

回复：回复: flink整合hbase

hi, sink是自己的,我还没发现flink有原生的hbase的sink,正在研究. 在2020年6月5日 09:06，xueaohui_...@163.com 写道： Sink 是自己的还是官方的。 xueaohui_...@163.com 发件人： liunaihua521 发送时间： 2020-06-05 00:27 收件人： user-zh@flink.apache.org 主题： flink整合hbase hi

回复： flink整合hbase

hi, 您能说下您本地的各组件版本吗?谢谢了 报错内容(不方便看附件有文本): org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: org.apache.flink.client.program.ProgramInvocationException: Job failed (JobID: cacd929120e6cefa79e1f8cca65e3d3f) at

Re:flink整合hbase

你好，可否提供一下报错信息呢？ 在 2020-06-05 00:27:42，liunaihua...@163.com 写道： >hi > 咨询下，想用flink1.10整合hbase，请问一下hadoop，hbase都需要什么版本？有哪位跑通过，可否提供下配置文件，我在本地测试，所有集群都正常启动，提交任务（flink读写hbase）总是报错，找不到某些类，是否跟版本有问题？ >提前谢谢了！ > >| | >liunaihua521 >| >| >邮箱：liunaihua...@163.com >| > >签名由 网易邮箱大师 定制

standalone模式下metaspace内存溢出

flink版本：1.10 standalone模式下，同一个batch作业提交多次后，taskmananger会metaspace内存溢出。 我用visualVm观察到，多次提交后，taskmanager的metaspace如下图： 虽然可以通过启动时调大jvm参数，但是这个应该是有问题的

Re: flink整合hbase

Hi, 可以贴下你的配置和异常栈吗？1.10的时候我本地也测试过，是ok的。 Best, Leonard Xu > 在 2020年6月5日，00:27，liunaihua...@163.com 写道： > > hi > 咨询下，想用flink1.10整合hbase，请问一下hadoop，hbase都需要什么版本？有哪位跑通过，可否提供下配置文件，我在本地测试，所有集群都正常启动，提交任务（flink读写hbase）总是报错，找不到某些类，是否跟版本有问题？ > 提前谢谢了！ > > | | > liunaihua521 > | > | >

Re: flink sql upsert模式写入mysql，es等key一定是groupby之后所有字段吗

感谢各位， 我先试试用Last_value这样的aggregate function绕过去。 @kcz 可能我表达不很清楚，具体描述一下遇到的具体场景就是：收到原始的数据，去mysql或者es里做维表关联，然后再以upsert的模式将结果写回mysql或es。 举个例子来说，我想按id为key更新整行数据（比如还有个字段amount是个随机的double类型值） select id， amount， …… groupby id； 这样子不行，必须 select id， amount， … groupby id，amount，……；

回复: flink整合hbase

Sink 是自己的还是官方的。 xueaohui_...@163.com 发件人： liunaihua521 发送时间： 2020-06-05 00:27 收件人： user-zh@flink.apache.org 主题： flink整合hbase hi 咨询下，想用flink1.10整合hbase，请问一下hadoop，hbase都需要什么版本？有哪位跑通过，可否提供下配置文件，我在本地测试，所有集群都正常启动，提交任务（flink读写hbase）总是报错，找不到某些类，是否跟版本有问题？ 提前谢谢了！ | | liunaihua521 | |

回复：flink sql upsert模式写入mysql，es等key一定是groupby之后所有字段吗

我大概get到你要说的需求，select那些其实是明细数据？但是没有跟聚合的数据拆开，所以才出现这种情况吧？ -- 原始邮件 -- 发件人: Leonard Xu https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/streaming/query_configuration.html#idle-state-retention-time

FlinkKinesisProducer sends data to only 1 shard. Is it because I don't have "AggregationEnabled" set to false ?

Hi, My FlinkKinesisProducer sends data to only 1 shard. Is it because I don't have "AggregationEnabled" set to false ? flink_connector_kinesis_2.11 : flink version 1.9.1 //Setup Kinesis Producer Properties kinesisProducerConfig = new Properties();

Flink sql nested elements

We have Avro schema that contains nested structure and when querying using Flink SQL, we are getting below error. Exception in thread "main" java.lang.AssertionError at org.apache.calcite.sql.parser.SqlParserPos.sum_(SqlParserPos.java:236) at

flink1.9 stream job的异常日志

附件是错误日志 我感觉看到错误日志之后，没有什么调查方向，应该怎么调查呢。2020-06-04 13:19:15,590 INFO org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction - FlinkKafkaProducer 7/12 - checkpoint 3425 complete, committing transaction TransactionHolder{handle=KafkaTransactionState [transactionalId=null,

Stopping a job

Hi: I am running a job which consumes data from Kinesis and send data to another Kinesis queue.  I am using an older version of Flink (1.6), and when I try to stop the job I get an exception  Caused by: java.util.concurrent.ExecutionException:

Creating TableSchema from the Avro Schema

Hi, In Flink 1.9, we have option to create the TableSchema form TypeInformation. We have used below. TypeInformation typeInfo = AvroSchemaConverter.convertToTypeInfo(schema); TableSchema tableSchema = TableSchema.fromTypeInfo(typeInfo); However TableSchema's fromTypeInfo method is deprecated

flink整合hbase

hi 咨询下，想用flink1.10整合hbase，请问一下hadoop，hbase都需要什么版本？有哪位跑通过，可否提供下配置文件，我在本地测试，所有集群都正常启动，提交任务（flink读写hbase）总是报错，找不到某些类，是否跟版本有问题？ 提前谢谢了！ | | liunaihua521 | | 邮箱：liunaihua...@163.com | 签名由 网易邮箱大师 定制

SQL Expression to Flink FilterFunction?

Hi there, I have a full DataStream pipeline (i.e. no upper level APIs like Table or SQL are involved). In the mid of the pipeline, I now need to filter on a SQL WHERE expression, e.g. "user= 'pitter' AND age > 10", which can include REGEXP and arbitrarily nested AND/OR constructs. I was

Avro Arrat type validation error

Hi, Avro schema contains Array type and we created TableSchema out of the AvroSchema and created a table in catalog. In the catalog, this specific filed type shown as ARRAY. We are using AvroRowDeserializationSchema with the connector and returnType of TableSource showing Array mapped to

Re: Native K8S not creating TMs

Thanks! I do not see any pods of the form `flink-taskmanager-1-1`, so I tried the exec suggestion. The logs are attached below. Is there a quick RBAC check I could perform? I followed the command on the docs page linked (kubectl create clusterrolebinding flink-role-binding-default

Re: Flink SQL 子查询优化问题

hi Yichao, 最好的解决方式是在code gen的时候支持表达式复用，后续会慢慢加上这些优化。 Best, Godfrey 1048262223 <1048262...@qq.com> 于2020年6月4日周四 下午9:52写道： > Hi > > > 感谢答复，我明天使用下这种方式 > 社区也有同学提供了udf使用udtf代替的方法 > 我会尝试使用上述两种方式实现下 > > > 还有想问下之后有对这部分做优化的计划吗？ > > > Best, > Yichao Yang > > > > > > -- 原始邮件

回复：Flink SQL 子查询优化问题

Hi 感谢答复，我明天使用下这种方式 社区也有同学提供了udf使用udtf代替的方法 我会尝试使用上述两种方式实现下 还有想问下之后有对这部分做优化的计划吗？ Best, Yichao Yang -- 原始邮件 -- 发件人: godfrey he

Re: Window Function use case;

If you input data already contains both the SensorID and FactoryID, why would the following not be sufficient? DataStream sensorEvents = ...; sensorEvents .filter(sensorEvent -> sensorEvent.Status.equals("alerte")) .map(sensorEvent -> sensorEvent.FactoryID) .addSink() If the problem is that

Re: flink sql upsert模式写入mysql，es等key一定是groupby之后所有字段吗

hi yihan, 如 Leonard 所说，你可以考虑使用 first_value, last_value 等聚合函数和赛选其他字段。 1.11开始支持ddl定义pk信息， 如果id在source表中也是pk字段，可以直接定义， planner会利用该信息传递pk到sink表。 Bests， Godfrey Leonard Xu 于2020年6月4日周四 下午9:01写道： > Hi, > > > 但这样子不能通过calcite的sqlvalidation，select后面不能有非聚合项， > > select后费聚合值可以通过max()或sum()来取，因为已经按照key

Re: 关于flinksql 与维表mysql的关联问题

hi 可以考虑使用 temporal table join [1] Best, Godfrey [1] https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/streaming/joins.html#join-with-a-temporal-table 小屁孩 <932460...@qq.com> 于2020年6月4日周四 下午5:51写道： > 您的意思是open 全量预加载吗？我目前的逻辑是自己自定义的source 广播出去 > 这是我的source > > > import

Re: Flink SQL 子查询优化问题

hi Yichao, 目前 planner 会 try best 的将两个相邻的 project 节点合并在一起，除非两个project被分开。 就像你上面的那种做法。但是加一个group by的执行代价比较高。 对于blink planner 而言，*有一个绕的办法*，可以在子查询的结果加一个print sink（可以ignore输出）， 利用多sink的优化特性，将两个project分开，从而阻止优化器将两个project合并。 Best, Godfrey 1048262223 <1048262...@qq.com> 于2020年6月4日周四 下午4:56写道： > Hi

Re: flink sql upsert模式写入mysql，es等key一定是groupby之后所有字段吗

Hi, > 但这样子不能通过calcite的sqlvalidation，select后面不能有非聚合项， select后费聚合值可以通过max()或sum()来取，因为已经按照key group by了，所以取出来的非聚合值只能有一条， > 这样子key的state无限增长(比如说amount是一个随机的double数)，job跑不久就会fail掉。 State 可以配置ttl的，过期清理参考[1] 另外，即将发布的1.11中，支持在jdbc table 上定义primary key， 不用强制要求写upsert 的query，文档正在撰写中[2] Best, Leonard

Window Function use case;

Hello guys, I have a use case, where I am receiving data from sensors about their status (Normal or Alerte), {SensorID:"1", FactoryID:"1", Status:"Normal" ..}, a factory can contain a lot of sensors, so what I want to do is, if the status of one sensor in a factory, is Alerte I want to raise an

Re: Suggestions for using both broadcast sync and conditional async-io

Thanks for the response, I had not seen the state processor API, somehow I missed that. Regarding your second point, this is basically an ID mapping service so I need the ID's persisted in the DynamoDB (or indeed any other external store) so that other applications may also use the 'mapped'

Re: Re： flink数据sink到mysql 是事务处理

您好： 我是在样的情况，flink处理完数据后数据被我组织成一个3元组，第一个元素代表的是对这个元素的操作(插入还是删除)，第二个是我的user_ID,第三个是一个list里面有多个实体； 数据要插入或者删除，因为list的大小不确定所以在操作mysql的时候开起了事务保证这一批次的数据幂等，但是程序运行一段时间后出现死锁，但是不影响数据，所以发邮件想知道大家 是怎么处理的，结果这么多天才有回复讨论； 伪代码如下： public RdsOperaterSink2(String url, String name, String password)

Re: pyflink下mysql数据更新操作

Hi, 小学生 我的理解是key如果是mysql主键的话，你正常sink进去，那条key=1,value=3的数据就会覆盖掉key=1,value=2的数据 Best, Xingbo 小学生 <201782...@qq.com> 于2020年6月4日周四 下午6:05写道： > 您好，第一个问题您说的这样我理解，但是我的问题是比如先前在mysql表已经有一条key=1,value=2的数据，但是接着我要按key=1时，value=3去覆盖掉key=1,value=2的数据，也就是mysql > 命令中update table **SET value=3 WHERE key=1;

??????yarn-session??job????????????????????????

-n 1.10 ---- ??:""

??????flink sql upsert????????mysql??es??key??????groupby??????????????

Hi ??key https://blog.csdn.net/lp284558195/article/details/104609739 https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/streaming/query_configuration.html#:~:text=The%20minimum%20idle%20state%20retention,kept%20before%20it%20is%20removed. Best, Yichao

yarn-session多job的模式下是动态申请资源吗

你好。我目前用的是flink-1.7.2版本，使用的是yarn-session多job的模式。启动yarn-session之后，在flink ui上并没有看到申请配置的资源，而是提交flink job的时候才申请的container。当提交多个job时，yarn-session会按照job需要的slot数量动态申请资源。 比如，启动yarn-session的配置为: ./bin/yarn-session.sh -tm 8192 -s 2 �Cn 4 提交一个job需要的资源是 8个slot，那么yarn-session会申请4个taskmanager来运行该job。

flink sql upsert模式写入mysql，es等key一定是groupby之后所有字段吗

举个例子比如我想要 insert into mysql_sink select ID， amount， ……… from source groupby ID； 这里就是想按照id为key，在数据库里更新这个id对应的amount等其他值。 但这样子不能通过calcite的sqlvalidation，select后面不能有非聚合项，必须在groupby后面加上所有select后面的项。 但这样带来一个问题，这样子key的state无限增长(比如说amount是一个随机的double数)，job跑不久就会fail掉。

flink sql ddl能设置ttl吗？

各位好， 我有一个需求是要维表关联mysql数据，再upsert写入mysql，就是个丰富字段的简单需求。 现在有两个问题： 1.为了实现upsert插入，我试下来必须使用groupby指定key，但是select内又有很多非聚合项，所以没办法在groupby后面加上所有那些非聚合项，否则通不过calcite validation。 2.现在遇到了一个问题，跑一段时间会抱full GC，我估计是因为groupby导致的状态不清理。 请问flink sql ddl能通过类似SET 语句来设定ttl吗，搜了官方文档只有提到table

Re: pyflink??mysql????????????

??mysqlkey=1,value=2??key=1value=3key=1,value=2??mysql ??update table **SET value=3 WHERE key=1;

Re: pyflink下mysql数据更新操作

Hi, 小学生 1. 关于问题1，这个是支持的，你通过TableEnvrironment的sql_update就行 例如： s_env = StreamExecutionEnvironment.get_execution_environment() s_env.set_parallelism(1) st_env = StreamTableEnvironment.create(s_env) mysql_source_ddl = """ CREATE TABLE buybacksource ( a VARCHAR, b VARCHAR, c VARCHAR, proctime AS

??????????flinksql ??????mysql??????????

??open source source import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.functions.source.RichSourceFunction; import java.sql.Connection; import java.sql.PreparedStatement; import

Re:?????? flink 1.9 ????????????????

?? json??kafka??flink ??json??table?? json ?? 2020-06-03 16:31:15??"star" <3149768...@qq.com> ?? > > >??select

Re:????flinksql ??????mysql??????????

open ?? 2020-06-04 14:15:05??"??" <932460...@qq.com> ?? >dear?? ??,flinksql??mysql >mysql??

Batch 模式 Table API 增加cache算子

hi all, 最初blink分支上有对batch模式下的table cache操作, 后续会merge到flink上来么？ -- Best, Jun Su

Re: flink1.10整合hbase测试遇到的问题

可以尝试把依赖包 shaded 到你的 jar 包里，保证依赖的完整 Best Weihua Hu > 2020年6月3日 22:52，liunaihua521 写道： > > > > - 转发邮件信息 - > > 发件人： liunaihua521 > 发送日期： 2020年6月3日 22:18 > 发送至： user-zh-i...@flink.apache.org   > 、

Flink SQL ??????????????

Hi all ?? Flink ??1.10 Planner??old planner/ blink planner ??Flink SQL source??Tuple2.of(1, "{\"name\": \"a\"}"); query??select a.id, a.name, a.name from (select id, body_json_to_map(name) as name from data) a sink??print udf??body_json_to_map

pyflink??mysql????????????

1.pyflink sql??update1.10.0 2.pyflink??Table

Re: pyflink ????????????????

get??udf??

Re: Getting Window information from coGroup functin

However, You can evaluate your time window, get the information of window of which an event is a part from the context of processFunction or any RichFunction you are passing the events to. So, on each event arrival you will be able to check which window the element is part of and from Window

Re: Getting Window information from coGroup functin

I think here apply function would receive only the events but not necessarily a complete window at same time. From: Dawid Wysakowicz Sent: Thursday, June 04, 2020 13:39 To: Sudan S; user@flink.apache.org Cc: Aljoscha Krettek Subject: Re: Getting Window

Re: Getting Window information from coGroup functin

I am afraid there is no way to do that. At least I could not think of a way to do it. Maybe @aljoscha cc'ed could help here. On 29/05/2020 13:25, Sudan S wrote: > Hi, > > I have a usecase where i want to join two streams. I am using coGroup > for this > > KeyBuilder leftKey = new >

Re: pyflink 嵌套使用函数出错

Hi, 小学生 我稍微修改了一下你的code(你的from_elements那样写按理说就没法运行) code是能够正确运行的，你可以参考一下，你去掉的是不是有问题，或者你把你修改后的代码贴上来，再一起看看 from pyflink.table import StreamTableEnvironment, DataTypes, EnvironmentSettings, TableConfig, \ BatchTableEnvironment from pyflink.table.descriptors import Schema, OldCsv, FileSystem from

Re: pyflink ????????????????

??

Re: pyflink 嵌套使用函数出错

Hi, 小学生。 把函数get的标签udf给去掉，它只是普通的Python函数，不要加上@udf，加上之后就不是python的函数了。只有Python的UDF你才要加上@udf Best, Xingbo 小学生 <201782...@qq.com> 于2020年6月4日周四 下午2:46写道： > 各位大佬好，初学pyflink,有一个问题需要帮忙解决下。 > > > 代码为： > from pyflink.table import StreamTableEnvironment, DataTypes, >

pyflink ????????????????

pyflink,?? from pyflink.table import StreamTableEnvironment, DataTypes, EnvironmentSettings,TableConfig,BatchTableEnvironment from pyflink.table.descriptors import Schema, OldCsv, FileSystem from pyflink.table.udf import udf from

??????????flinksql ??????mysql??????????

Hi jobjobsavepoint?? Best, Yichao Yang ---- ??:"??"<932460...@qq.com; :2020??6??4??(??) 2:15 ??:"user-zh"

??????????flinksql ??????mysql??????????

Hi?? JOIN??WaterMark??maxoutoforderness ---- ??:"??"<932460...@qq.com; :2020??6??4??(??) 2:15 ??:"user-zh"

????flinksql ??????mysql??????????

dear?? ??,flinksql??mysql mysql?? mysql??mysqlmysql

Re: Suggestions for using both broadcast sync and conditional async-io

Hi, For the initial DB fetch and state bootstrapping: That's exactly what the State Processor API is for, have you looked at that already? It currently does support bootstrapping broadcast state [1], so that should be good news for you. As a side note, I may be missing something, is broadcast