Re: flink sql 写 hive分区表失败

Hi, >|select * from dwdCatalog.dwd.t1 where `p_year` = 2020 and `p_month` = > 5 应该是 select * 会把分区字段一起带出来吧，你字段就不匹配了，select里加上你需要的字段吧 祝好， Leonard Xu > 在 2020年5月28日，12:57，Zhou Zach 写道： > > org.apache.flink.client.program.ProgramInvocationException: The main method > caused an error:

flink sql 写 hive分区表失败

org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Field types of query result and registered TableSink dwdCatalog.dwd.t1_copy do not match. Query schema: [id: BIGINT, name: STRING, p_year: INT, p_month: INT, EXPR$4: INT NOT NULL, EXPR$5: INT NOT NULL]

回复： flink 如何自定义connector

Hi, 在sql-gateway里面，类加载默认是子类优先，每次提交任务都是独立的tableEnvironment，独立的classloader，因此不存在这个问题。 Best, Xinghalo

回复： flink 如何自定义connector

可能的确会存在这个问题。在添加依赖时注意不要将 flink provided 的包打包进去就可以。也可以通过设置 parent-first 来解决这个问题。 在2020年05月28日 11:03，forideal 写道： Hi 111, 关于第二点： `2 配置FLINK_HOME环境变量，自定义的connector放在FLINK_HOME/lib下` 这么做是不是存在如下 `X can't be cast to X ` 隐患 因为把 Connector 放在 lib 中，会有 classloader 的问题，直接的现象就是 X can't be cast to X

Re:回复： flink 如何自定义connector

Hi 111, 关于第二点： `2 配置FLINK_HOME环境变量，自定义的connector放在FLINK_HOME/lib下` 这么做是不是存在如下 `X can't be cast to X ` 隐患 因为把 Connector 放在 lib 中，会有 classloader 的问题，直接的现象就是 X can't be cast to X 问题[1]。当然这只是说可能会发生。比如，我们把 usercode 代码放入 Flink lib 我们会发现，当我们使用 jar 包上传的方式运行任务时，jar 中也包含 lib 中的代码会触发这样的问题。 [1]

Re: 向flink push代码

Flink 的特点就是快（x） Best, tison. 宇张 于2020年5月28日周四 上午10:56写道： > 感谢大佬们，我看到 Leonard Xu大佬已经关注了FLINK-17991 > 这个，好快的响应速度 > > On Thu, May 28, 2020 at 10:25 AM Leonard Xu wrote: > > > Hi, > > Yangze 贴了官方教程，也可以看下 Jark 的博客[1]，中文的看起来会快一些。 > > > > Best, >

Re: 向flink push代码

感谢大佬们，我看到 Leonard Xu大佬已经关注了FLINK-17991 这个，好快的响应速度 On Thu, May 28, 2020 at 10:25 AM Leonard Xu wrote: > Hi, > Yangze 贴了官方教程，也可以看下 Jark 的博客[1]，中文的看起来会快一些。 > > Best, > Leonard Xu > [1]

Re: Table API 有 类似于 SQL API 的 TO_TIMESTAMP 函数吗

Get it, thanks Leonard Xu 于2020年5月28日周四 上午10:34写道： > > > 我原以为 Table API 会比 SQL API 自由度大一些，毕竟可以代码层面定制 > > > Table API 确实是要灵活些，只是最近两个版本中，SQL模块社区的主要精力在搞DDL，DDL可以降低用户门槛，提升易用性，Table > API的迭代稍微慢了些， > 我理解1.12中应该会完善Descriptor API，这也是比较重要的用户入口，目前建议优先使用DDL。 > > Best, > Leonard Xu > > > 在

Re: Table API 有 类似于 SQL API 的 TO_TIMESTAMP 函数吗

> 我原以为 Table API 会比 SQL API 自由度大一些，毕竟可以代码层面定制 Table API 确实是要灵活些，只是最近两个版本中，SQL模块社区的主要精力在搞DDL，DDL可以降低用户门槛，提升易用性，Table API的迭代稍微慢了些， 我理解1.12中应该会完善Descriptor API，这也是比较重要的用户入口，目前建议优先使用DDL。 Best, Leonard Xu > 在 2020年5月28日，10:23，macia kk 写道： > > 好的，谢谢， > > 放弃治疗，我先尝试DDL，先把 job 跑通，我原以为 Table API

Re: 向flink push代码

Hi, Yangze 贴了官方教程，也可以看下 Jark 的博客[1]，中文的看起来会快一些。 Best, Leonard Xu [1] https://wuchong.me/blog/2019/02/12/how-to-become-apache-committer/ > 在 2020年5月28日，10:18，Yangze Guo 写道： > > 您好，社区的贡献代码教程[1]。 > > Tips:

Re: Table API 有 类似于 SQL API 的 TO_TIMESTAMP 函数吗

好的，谢谢， 放弃治疗，我先尝试DDL，先把 job 跑通，我原以为 Table API 会比 SQL API 自由度大一些，毕竟可以代码层面定制 Leonard Xu 于2020年5月28日周四 上午10:17写道： > Hi， > 我看了下Descriptor的代码，如果数据源是Kafka应该有地方绕，很绕, 你可以试下，建议使用DDL。 > > Best > Leonard Xu > > [1] >

Re: 向flink push代码

您好，社区的贡献代码教程[1]。 Tips: 如果不是hotfix，请提交前确认已经在ticket下达成一致并且有committer将其assign给您。 [1] https://flink.apache.org/zh/contributing/contribute-code.html Best, Yangze Guo On Thu, May 28, 2020 at 10:15 AM 宇张 wrote: > > 找打了教程了 > > > On Thu, May 28, 2020 at 10:10 AM 宇张 wrote: > > > hi， > >

Re: Table API 有 类似于 SQL API 的 TO_TIMESTAMP 函数吗

Hi， 我看了下Descriptor的代码，如果数据源是Kafka应该有地方绕，很绕, 你可以试下，建议使用DDL。 Best Leonard Xu [1] https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/table/descriptors/Kafka.java#L111

回复： flink 如何自定义connector

Hi, 想要在sqlgateway里面使用，那么可以看看下面几个条件： 1 满足SPI的要求，能让flink自动发现实现类 2 配置FLINK_HOME环境变量，自定义的connector放在FLINK_HOME/lib下 3 如果与Hive集成，使用hivecatalog，那么先要注册表 这样就可以使用了。 Best, Xinghalo

Re: 向flink push代码

找打了教程了 On Thu, May 28, 2020 at 10:10 AM 宇张 wrote: > hi， > 请问给flink社区push代码的流程是怎么样的哪，有没有小白教程啊。。 >

回复： flink 如何自定义connector

Hi, 随便开一个jdbc connector之类的，模仿造一个就行： 1 需要有Service loader的描述符文件：resources下面需要有META-INFO.services，里面有TableFactory的实现类声明 2 创建对应的TableFactory实现类，根据source sink实现不同的接口，返回对应的TableSource或者TableSink 3 如果是TableSource，根据剪枝、lookup等特性实现不同的接口 4 如果是TableSink，根据upsert、append、retract实现不同的接口 总之，开一个connector源码，抄一下就行。

向flink push代码

hi， 请问给flink社区push代码的流程是怎么样的哪，有没有小白教程啊。。

Re: flink 如何自定义connector

hello 现在已经定义了一个tablesource，可以通过 batchEnv.registerTableSource 这个注册 并查询数据，但是如何在sqlgateway 中配置呢？ Leonard Xu 于2020年5月28日周四 上午9:32写道： > Hi， > 可以参考现有的connector，如hbase，jdbc，结合[1]实现自定义connector。 > > > 祝好， > Leonard Xu > [1] >

Re: 回复：flink 1.9 1.10 on yarn在cdh上怎么搭建一个客户端

确实，你只要配置好CDH的HADOOP_CONF环境变量，同时下载开源的Hadoop版本（和CDH版本相同）放到flink lib下，就可以访问CDH yarn，提交作业！ 目前我这边是CDH 5.16.1，Flink 1.10，提交Flink on yarn是没问题，任务运行也没问题，还可以使用Flink on hive！ flink-shaded-hadoop-2-uber-2.6.5-10.0.jar 发件人： 111 发送时间： 2020-05-28 09:13 收件人： user-zh@flink.apache.org

Re: flink 如何自定义connector

Hi， 可以参考现有的connector，如hbase，jdbc，结合[1]实现自定义connector。 祝好， Leonard Xu [1] https://ci.apache.org/projects/flink/flink-docs-master/dev/table/sourceSinks.html > 在 2020年5月28日，09:16，Peihui He 写道： > > hello > >

flink 如何自定义connector

hello 请教大家，flink 1.10中如何自定义coonnecter，然后注册到flink sql gateway，使得可以执行sql的操作呢？ best wish

回复：flink 1.9 1.10 on yarn在cdh上怎么搭建一个客户端

Hi， 一般只要你有yarn环境，在任意一台机器上下载flink安装包，配一下HADOOP_CONF环境变量就可以使用。 如果是session模式：可以使用Yarn-session.sh启动yarn session集群，然后通过flink run xxx 提交程序。 如果是per job模式：直接使用flink run即可。 best， Xinghalo

flink 1.9 1.10 on yarn在cdh上怎么搭建一个客户端

flink 1.9 1.10 在cdh上怎么搭建一个客户端。 我需要一个客户端启动flink on yan. 1.7版本 是正常的。 但是1.9 和1.10 启动了on yarn 任务。我的环境是cdh hadoop。 谢谢回答

Table API 有 类似于 SQL API 的 TO_TIMESTAMP 函数吗

Hi 各位大佬 .field("event_time", TIMESTAMP()).rowtime( new Rowtime() .timestampsFromField("maxwell_ts") .watermarksPeriodicBounded(6) ) 我这个 maxwell_ts 是 milliseconds ，直接这么用会报错： Type TIMESTAMP(6) of table field ‘event_time’ does not match with the

Re: Flink sql 跨库

问题解决了就好， > 找到原因了，flink 把year 当成关键字了 YEAR在时间相关类型中会使用，在 FLINK SQL 是关键字的，在一些DB里也是关键字，使用时需要转义，类似的还有DAY, MONTH等[1]。 [1] https://ci.apache.org/projects/flink/flink-docs-master/dev/table/types.html#interval-year-to-month

Re:Re: Re: Re: Re: Flink sql 跨库

好的，感谢指点 在 2020-05-27 19:33:42，"Rui Li" 写道： >你是想要调试HiveCatalog的代码么？可以参考flink里的测试用例，我们有的测试是用embedded模式做的（比如HiveCatalogHiveMetadataTest），有些测试是单独起一个HMS进程（比如TableEnvHiveConnectorTest）。 > >On Wed, May 27, 2020 at 7:27 PM Zhou Zach wrote: > >> 是的，发现了，感谢指点。请教下，用intellij >>

Re: Re: Re: Re: Flink sql 跨库

你是想要调试HiveCatalog的代码么？可以参考flink里的测试用例，我们有的测试是用embedded模式做的（比如HiveCatalogHiveMetadataTest），有些测试是单独起一个HMS进程（比如TableEnvHiveConnectorTest）。 On Wed, May 27, 2020 at 7:27 PM Zhou Zach wrote: > 是的，发现了，感谢指点。请教下，用intellij > idea调试，你是在本地调试吗，那样的话，要在本地搭建个hadoop集群吗，至少要搭建个本地的hive吧，还是直接用intellij >

Re:Re: Re: Re: Flink sql 跨库

是的，发现了，感谢指点。请教下，用intellij idea调试，你是在本地调试吗，那样的话，要在本地搭建个hadoop集群吗，至少要搭建个本地的hive吧，还是直接用intellij idea连接远程，如果集群在阿里云上，是不是要另外开端口的 在 2020-05-27 19:19:58，"Rui Li" 写道： >year在calcite里是保留关键字，你用`year`试试呢 > >On Wed, May 27, 2020 at 7:09 PM Zhou Zach wrote: > >> The program finished with the

Re:Re:Re: Re: Flink sql 跨库

找到原因了，flink 把year 当成关键字了 At 2020-05-27 19:09:43, "Zhou Zach" wrote: >The program finished with the following exception: > > >org.apache.flink.client.program.ProgramInvocationException: The main method >caused an error: SQL parse failed. Encountered "year =" at line 4, column

Re: Re: Re: Flink sql 跨库

year在calcite里是保留关键字，你用`year`试试呢 On Wed, May 27, 2020 at 7:09 PM Zhou Zach wrote: > The program finished with the following exception: > > > org.apache.flink.client.program.ProgramInvocationException: The main > method caused an error: SQL parse failed. Encountered "year =" at line 4, > column

Re:Re: Re: Flink sql 跨库

The program finished with the following exception: org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: SQL parse failed. Encountered "year =" at line 4, column 51. Was expecting one of: "ARRAY" ... "CASE" ... "CURRENT" ...

Re: Re: Flink sql 跨库

读hive分区表报的什么错啊，把stacktrace贴一下？ On Wed, May 27, 2020 at 6:08 PM Zhou Zach wrote: > > > hive partition table: > > > 1CREATE TABLE `dwd.bill`( > 2 `id` bigint, > 3 `gid` bigint, > 4 `count` bigint, > 5 `price` bigint, > 6 `srcuid` bigint, > 7 `srcnickname` string, > 8 `srcleftmoney` bigint,

Re: Flink SQL FIRST_VALUE 是append 模式吗？

你的理解是对的。你可以尝试下用time windowed join[1]，这个不管是什么join类型，结果都是append的。 [1] https://ci.apache.org/projects/flink/flink-docs-master/dev/table/sql/queries.html#joins macia kk 于2020年5月27日周三 下午6:56写道： > 感谢 Benchao > > 原来如此，我之前用的是 spark structured Streming, 可能模式跟Flink不太一样，它会缓存没有 join 到的数据，直到 > watermark

Re: Flink SQL FIRST_VALUE 是append 模式吗？

感谢 Benchao 原来如此，我之前用的是 spark structured Streming, 可能模式跟Flink不太一样，它会缓存没有 join 到的数据，直到 watermark 结束之后才 emit Flink 新的数据进来跟右边的缓存数据 join, 没有 join 到先发 null，但是这个数据还会缓存, 后边右边如果有新的数据可以 join 到左边已经发出去的这条数据，会产生 retract. (我的理解) 那我这种情况有别的解决方案吗？因为我的 Sink (Kafka) 下游是 Druid， 数据会直接 index 后作为查询，不支持 retract 场景。

Re: Flink SQL FIRST_VALUE 是append 模式吗？

产生retract消息的场景有很多，暂时还没有一篇文档来介绍这个，我大概列举几个典型的场景吧： 1. regular group by，因为聚合的结果是实时下发的，所以更新了聚合结果就会retract老的聚合结果 2. 非inner/anti 的join（不包括time interval join)，这种原因是如果当前join不上，会发送null，但是后面可能对面可能又会有数据进来，导致下发的null需要被retract 3. 取latest的去重 4. topn，排名变化需要更新结果 5. window + emit，提前emit的结果需要retract来更新 macia kk

Re: Flink SQL FIRST_VALUE 是append 模式吗？

感谢 Benchao 和 Leonard 的回复 我理解错误的地方在于，我以为 Left join 是 append 模式的，对于左边的数据来说，join 上一条就会 emit 出去，但是什么情况下会产生 react 消息呢？ Leonard Xu 于2020年5月27日周三 下午3:50写道： > Hi > Kafka 的 sink 是 appendStreamSink，所以query只能是append的, 因为Kafka 的 > sink无法处理retract消息。 > 你SQL中的 left join

Re:Re: Flink sql 跨库

hive partition table: 1CREATE TABLE `dwd.bill`( 2 `id` bigint, 3 `gid` bigint, 4 `count` bigint, 5 `price` bigint, 6 `srcuid` bigint, 7 `srcnickname` string, 8 `srcleftmoney` bigint, 9 `srcwealth` bigint, 10 `srccredit` decimal(10,0), 11 `dstnickname` string, 12 `dstuid`

Re: Flink sql 跨库

Flink 支持hive分区表的，看你在另外一个邮件里贴了，你能把你的hive表和query在邮件里贴下吗？ 祝好 Leonard Xu > 在 2020年5月27日，17:40，Zhou Zach 写道： > > > > > 感谢回复，表名前加上Catalog和db前缀可以成功访问了。 > 现在遇到个问题，flink 读hive 分区表时，如果where子句用分区键，比如year过滤就会报错，用表中其他字段过滤是没问题的，是flink 不支持 > hive分区表，还是哪个地方没设置对 > > > > > > > > > > > > > > >

Re: flink-python 配置文件问题

你好，使用-pyfs指定的python文件会在运行时将python文件加入到PYTHONPATH下面，不会加到classpath下面。 guaishushu1...@163.com 于2020年5月27日周三 下午5:50写道： > 使用命令-pysf 加载python配置文件 发现classpath并没有这个文件 > > -- > guaishushu1...@163.com >

flink-python 配置文件问题

使用命令-pysf 加载python配置文件 发现classpath并没有这个文件 guaishushu1...@163.com

Re:Re: Flink sql 跨库

感谢回复，表名前加上Catalog和db前缀可以成功访问了。 现在遇到个问题，flink 读hive 分区表时，如果where子句用分区键，比如year过滤就会报错，用表中其他字段过滤是没问题的，是flink 不支持 hive分区表，还是哪个地方没设置对 在 2020-05-27 17:33:11，"Leonard Xu" 写道： >Hi， >> 因为一个HiveCatalog只能关联一个库 >一个Catalog是可以关联到多个db的，不同catalog，不同db中表都可以访问的. > >Flink SQL> show catalogs;

Re: Flink sql 跨库

Hi， > 因为一个HiveCatalog只能关联一个库 一个Catalog是可以关联到多个db的，不同catalog，不同db中表都可以访问的. Flink SQL> show catalogs; default_catalog myhive Flink SQL> use catalog myhive; Flink SQL> show databases; default hive_test hive_test1 Flink SQL> select * from hive_test.db2_table union select * from

Re: Flink SQL FIRST_VALUE 是append 模式吗？

Hi Kafka 的 sink 是 appendStreamSink，所以query只能是append的, 因为Kafka 的 sink无法处理retract消息。 你SQL中的 left join 是个双流join，会产生retract消息，编译时发现query和sink不匹配会报你贴的错误， 通常看到的业务场景是写入kafka的数据是不用去更新的，kafka主要用作一个消息队列，如果是需要更新 结果的数据，一般是存到db里，jdbc，hbase, es 这几个 connector 的sink都是upsertSink。 祝好, Leonard Xu > 在

Re: 回复：全局state

不会，在第二次keyby的时候，根据id-name 做key， 然后做reduce， 把重复的数据中的value，根据time去做判断，取time最小值的value，就可以去重了 a773807...@gmail.com 发件人： star 发送时间： 2020-05-27 15:45 收件人： user-zh 主题： 回复：回复：全局state 感谢您的建议，但是这样数据会重复，翻倍。目前看好像只能依赖外部存储了 发自我的iPhone -- 原始邮件 -- 发件人:

回复：回复：全局state

感谢您的建议，但是这样数据会重复，翻倍。目前看好像只能依赖外部存储了 发自我的iPhone -- 原始邮件 -- 发件人: a773807...@gmail.com https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/api_concepts.html#define-keys-using-key-selector-functions Best, tison. star <3149768...@qq.comgt;

回复：native kubernetes在不同kubernetes版本构建失败问题

感谢你的耐心解答～ | | a511955993 | | 邮箱：a511955...@163.com | 签名由 网易邮箱大师 定制 在2020年05月27日 14:17，Yang Wang 写道： 是的 不过后续我觉得也可以考虑把这个fix backport到1.10分支，在下一次发布的时候修复 目前master版本已经包含了这个fix，但是需要你自己来build镜像 于2020年5月27日周三 下午1:38写道： > hi，yang > > 使用的镜像是docker hub提供的1.10.1版本镜像。因此解法是 > 1. 等待1.11版本 > 2.

Re: native kubernetes在不同kubernetes版本构建失败问题

是的 不过后续我觉得也可以考虑把这个fix backport到1.10分支，在下一次发布的时候修复 目前master版本已经包含了这个fix，但是需要你自己来build镜像 于2020年5月27日周三 下午1:38写道： > hi，yang > > 使用的镜像是docker hub提供的1.10.1版本镜像。因此解法是 > 1. 等待1.11版本 > 2. 自行构建flink 1.10.1版本镜像，降低jdk版本？ > > Looking forward to your reply and help. > > Best > > > > > > | | > a511955993 >

回复：flink集群启动显示正在运行instance多,StandaloneSessionClusterEntrypoint启动后被kill

还是不行，可能是别的原因，昨天还好好的，今天不知道搞了什么就这样了。看了日志说resourcemanager拒绝连接。 ---原始邮件--- 发件人: Lijie Wang