【Announce】Zeppelin 0.10.0 is released, Flink on Zeppelin Improved
Hi Flink users, We (Zeppelin community) are very excited to announce Zeppelin 0.10.0 is officially released. In this version, we made several improvements on Flink interpreter. Here's the main features of Flink on Zeppelin: - Support multiple versions of Flink - Support multiple versions of Scala - Support multiple languages - Support multiple execution modes - Support Hive - Interactive development - Enhancement on Flink SQL - Multi-tenancy - Rest API Support Take a look at this document for more details: https://zeppelin.apache.org/docs/0.10.0/interpreter/flink.html The quickest way to try Flink on Zeppelin is via its docker image https://zeppelin.apache.org/docs/0.10.0/interpreter/flink.html#play-flink-in-zeppelin-docker Besides these, here’s one blog about how to run Flink sql cookbook on Zeppelin, https://medium.com/analytics-vidhya/learn-flink-sql-the-easy-way-d9d48a95ae57 The easy way to learn Flink Sql. Hope it would be helpful for you and welcome to join our community to discuss with others. http://zeppelin.apache.org/community.html -- Best Regards Jeff Zhang
Re: Flink使用SQL注册UDF未来有规划吗
https://www.yuque.com/jeffzhangjianfeng/gldg8w/dthfu2 Ada Luna 于2021年8月16日周一 下午2:26写道： > 目前注册UDF要通过Table API。 > 未来会通过SQL直接将UDF注册到上下文中吗？ > -- Best Regards Jeff Zhang
Re: flink sql 依赖隔离
Zeppelin 支持依赖的动态加载 https://www.yuque.com/jeffzhangjianfeng/gldg8w/rn6g1s https://www.yuque.com/jeffzhangjianfeng/gldg8w/dthfu2 Michael Ran 于2021年7月22日周四 下午8:07写道： > 通过任务进行隔离引用呗。你们美团已经是k8s了吧？ > 在 2021-07-05 14:06:53，"silence" 写道： > >请教大家目前flink sql有没有办法做到依赖隔离 > >比如connector，format，udf(这个最重要)等， > >很多情况下不同用户定义的组件会经常混用，会有很多的依赖冲突问题难以解决。 > >目前有没有办法做到udf的依赖隔离（比如不同的udf使用独立的jar和classloader），或者社区对此有没有什么规划 > -- Best Regards Jeff Zhang
Re: Re: flink sql平台多版本支持问题
如果不是native k8s的话，现在已经支持了，用remote模式就可以， https://www.yuque.com/jeffzhangjianfeng/gldg8w/engh3w native k8s的话，社区正在做，这是PR： https://github.com/apache/zeppelin/pull/4116 casel.chen 于2021年6月13日周日 上午9:39写道： > 嗯，flink on zeppelin也是一个不错的选择，只是因为现在flink on > zeppelin还不支持作业运行在kubernetes上，所以暂时还无法直接使用，未来支持后可以考虑引入。 > 谢谢大佬给的建议。 > > > > > > > > > > > > > > > > > > 在 2021-06-13 07:21:46，"Jeff Zhang" 写道： > >另外一个选择是flink on zeppelin，可以调用flink on zeppelin的rest api，把zeppelin当做是flink > >job server， zeppelin天然支持flink 1.10之后的所有版本。钉钉群：32803524 > > > >casel.chen 于2021年6月12日周六 下午5:56写道： > > > >> 需求背景： > >> 因为不同Flink版本SQL底层实现不相同，同一个带状态的Flink SQL作业目前跨版本升级社区还不支持。所以如果基于Flink > >> SQL开发实时计算平台的话，平台需要支持不同Flink版本。早期的Flink SQL作业用的是1.11，最新的Flink > >> SQL作业采用的是1.13开发的。 > >> > >> > >> 而让平台支持不同Flink版本，我能想到有三种实现方案： > >> > >> > >> 1. 平台直接调用 flink run 或 flink run-application 提交作业 > >> 优点：实现简单，每个flink版本都会带这个shell脚本 > >> > >> > 缺点：受限于脚本提供的功能，像语法检查、格式化sql、获取sql执行计划及sql结构和血缘这些“高级”功能没有在官方脚本中提供(能否完善下呢？），如果要使用的话得自己扩展实现（同样需要自己维护多版本） > >> > >> > >> 2. 平台调用 flink sql gateway 暴露的 restful api 提交作业 > >> 优点：实现简单，通过gateway屏蔽上层应用调用多版本flink api问题 > >> 缺点：需要额外扩展开发和部署gateway，目前暂不清楚社区对gateway的定位是什么？只是配合flink jdbc > >> driver使用么？这两项目目前活跃度都不高 > >> > >> > >> 3. 需要维护多个平台版本，像 flink-sql-connector-hive 支持多版本hive那样 > >> 优点：现有平台就是基于特定版本flink api开发的，改动小，需要抽取公用接口 > >> 缺点：需要“复制”平台多份实现以支持不同flink版本，开发和维护量比较大 > >> > >> > >> 目前，我暂时prefer方案2，可能不对，也可能还有更优的我没想到，欢迎各位大佬们给点建议。 > >> > >> > > > >-- > >Best Regards > > > >Jeff Zhang > -- Best Regards Jeff Zhang
Re: flink sql平台多版本支持问题
另外一个选择是flink on zeppelin，可以调用flink on zeppelin的rest api，把zeppelin当做是flink job server， zeppelin天然支持flink 1.10之后的所有版本。钉钉群：32803524 casel.chen 于2021年6月12日周六 下午5:56写道： > 需求背景： > 因为不同Flink版本SQL底层实现不相同，同一个带状态的Flink SQL作业目前跨版本升级社区还不支持。所以如果基于Flink > SQL开发实时计算平台的话，平台需要支持不同Flink版本。早期的Flink SQL作业用的是1.11，最新的Flink > SQL作业采用的是1.13开发的。 > > > 而让平台支持不同Flink版本，我能想到有三种实现方案： > > > 1. 平台直接调用 flink run 或 flink run-application 提交作业 > 优点：实现简单，每个flink版本都会带这个shell脚本 > > 缺点：受限于脚本提供的功能，像语法检查、格式化sql、获取sql执行计划及sql结构和血缘这些“高级”功能没有在官方脚本中提供(能否完善下呢？），如果要使用的话得自己扩展实现（同样需要自己维护多版本） > > > 2. 平台调用 flink sql gateway 暴露的 restful api 提交作业 > 优点：实现简单，通过gateway屏蔽上层应用调用多版本flink api问题 > 缺点：需要额外扩展开发和部署gateway，目前暂不清楚社区对gateway的定位是什么？只是配合flink jdbc > driver使用么？这两项目目前活跃度都不高 > > > 3. 需要维护多个平台版本，像 flink-sql-connector-hive 支持多版本hive那样 > 优点：现有平台就是基于特定版本flink api开发的，改动小，需要抽取公用接口 > 缺点：需要“复制”平台多份实现以支持不同flink版本，开发和维护量比较大 > > > 目前，我暂时prefer方案2，可能不对，也可能还有更优的我没想到，欢迎各位大佬们给点建议。 > > -- Best Regards Jeff Zhang
Re: 提交flink-sql 出现无法部署到yarn集群
看 yarn app log 张锴 于2021年4月14日周三 下午8:00写道： > 在用flink-sql的方式连接hive时 ，出现以下错误： > The program finished with the following exception: > > org.apache.flink.client.program.ProgramInvocationException: The main method > caused an error: Could not deploy Yarn job cluster. > at > > org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:366) > at > > org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:219) > at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114) > at > > org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:812) > at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:246) > at > org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1054) > at > > org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1132) > at java.security.AccessController.doPrivileged(Native Method) > at javax.security.auth.Subject.doAs(Subject.java:422) > at > > org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1688) > at > > org.apache.flink.runtime.security.contexts.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) > at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1132) > Caused by: org.apache.flink.client.deployment.ClusterDeploymentException: > Could not deploy Yarn job cluster. > at > > org.apache.flink.yarn.YarnClusterDescriptor.deployJobCluster(YarnClusterDescriptor.java:481) > at > > org.apache.flink.client.deployment.executors.AbstractJobClusterExecutor.execute(AbstractJobClusterExecutor.java:81) > at > > org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:1905) > at > > org.apache.flink.client.program.StreamContextEnvironment.executeAsync(StreamContextEnvironment.java:135) > at > > org.apache.flink.client.program.StreamContextEnvironment.execute(StreamContextEnvironment.java:76) > at > > org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1782) > at > > org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1765) > at com.erwan.flinksql.FlinkConnectHive.main(FlinkConnectHive.java:49) > at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) > at > > sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) > at > > sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) > at java.lang.reflect.Method.invoke(Method.java:498) > at > > org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:349) > ... 11 more > Caused by: > org.apache.flink.yarn.YarnClusterDescriptor$YarnDeploymentException: The > YARN application unexpectedly switched to state FAILED during deployment. > Diagnostics from YARN: Application application_1618298202025_0017 failed 1 > times (global limit =2; local limit is =1) due to AM Container for > appattempt_1618298202025_0017_01 exited with exitCode: 2 > Failing this attempt.Diagnostics: [2021-04-14 19:04:02.506]Exception from > container-launch. > Container id: container_e13_1618298202025_0017_01_01 > Exit code: 2。 > > 由于错误原因不明显，不好排查，也不确定是到底是哪里的问题，请问有什么办法能够定位问题。 > -- Best Regards Jeff Zhang
Re: Re: flink sql如何从远程加载jar包中的udf
Zeppelin 支持加载UDF jar的，可以参考下面的代码，不过架构上可能与你们的原有架构会有所差别 https://www.yuque.com/jeffzhangjianfeng/gldg8w/dthfu2#8iONE https://github.com/apache/zeppelin/blob/master/flink/interpreter/src/main/scala/org/apache/zeppelin/flink/FlinkScalaInterpreter.scala#L469 chenxyz 于2021年3月12日周五 上午9:42写道： > 目前这种方法不可行，在公司的平台化系统里提交flink任务，自己能掌控的只有代码这块。 > > > > > > > > > > > > > > > > > > 在 2021-03-11 16:39:24，"silence" 写道： > >启动时通过-C加到classpath里试试 > > > > > > > >-- > >Sent from: http://apache-flink.147419.n8.nabble.com/ > -- Best Regards Jeff Zhang
Re: flink-sql-gateway稳定性如何，可以在生产环境使用吗？
zeppelin 有 rest api 接口，https://www.yuque.com/jeffzhangjianfeng/gldg8w/pz2xoh jinsx 于2021年1月27日周三 下午2:30写道： > 如果使用zeppelin，zeppelin可以提供rpc接口吗 > > > > -- > Sent from: http://apache-flink.147419.n8.nabble.com/ > -- Best Regards Jeff Zhang
Re: 获取flinksql返回的查询结果
Sink 到内存里，然后你自己处理（print出来还是发送到web前端） 可以参考zeppelin源码 https://github.com/apache/zeppelin/tree/master/flink 黑色 于2021年1月18日周一 下午8:42写道： > 请教个问题，我想实现类似sql-cli里功能一样，或者zepplin里的我在页面上编写select * from ， > > > 在页面下面得到返回的结果显示，类似zepplin在开发区写sql,下面的输出区显示返回的结果，这个功能如何实现呢 > 各位大哥们帮忙看一看？ > > > 现在在flinksql上想看结果，还的定义一个with='print'，跑到ui页面上去看，太麻烦了 -- Best Regards Jeff Zhang
[Announce] Zeppelin 0.9.0 is released (Flink on Zeppelin)
Hi flink folks, I'd like to tell you that Zeppelin 0.9.0 is officially released, in this new version we made a big refactoring and improvement on flink support. It supports 3 major versions of flink (1.10, 1.11, 1.12) You can download zeppelin 0.9.0 here. http://zeppelin.apache.org/download.html And check the following 2 links for more details of how to use flink on zeppelin https://app.gitbook.com/@jeffzhang/s/flink-on-zeppelin/ http://zeppelin.apache.org/docs/0.9.0/interpreter/flink.html -- Best Regards Jeff Zhang
Re: 构建Flink任务提交平台时兼容低版本Flink的问题
可以参考 zeppelin的方法，zeppelin支持多个版本的flink （1.10, 1.11, 1.12) https://www.yuque.com/jeffzhangjianfeng/gldg8w/bam5y1 https://github.com/apache/zeppelin/tree/master/flink Shengnan YU 于2021年1月16日周六 上午8:15写道： > 请问各位大佬，目前我用Flink > 1.11的内部API实现了Flink任务提交的平台，但是由于1.11相对1.10大改了客户端的实现方案，导致低版本的Flink应用难以适配，请问大家有没有在构建平台时有兼容多版本Flink任务的好方案，并且能够方便快速升级迭代支持新的Flink版本，我现在想到的就是用命令行去做。 > -- Best Regards Jeff Zhang
Re: zeppelin+flink1.12问题
钉钉群里的新版本已经解决了，钉钉群号：32803524 赵一旦 于2020年12月15日周二 下午4:37写道： > 如题，zeppelin+flink1.12报错（select 1）。 > org.apache.zeppelin.interpreter.InterpreterException: > org.apache.zeppelin.interpreter.InterpreterException: > java.lang.NoSuchMethodError: > > org.apache.flink.api.common.ExecutionConfig.disableSysoutLogging()Lorg/apache/flink/api/common/ExecutionConfig; > at > > org.apache.zeppelin.interpreter.LazyOpenInterpreter.open(LazyOpenInterpreter.java:76) > at > > org.apache.zeppelin.interpreter.remote.RemoteInterpreterServer$InterpretJob.jobRun(RemoteInterpreterServer.java:760) > at > > org.apache.zeppelin.interpreter.remote.RemoteInterpreterServer$InterpretJob.jobRun(RemoteInterpreterServer.java:668) > at org.apache.zeppelin.scheduler.Job.run(Job.java:172) > at > > org.apache.zeppelin.scheduler.AbstractScheduler.runJob(AbstractScheduler.java:130) > at > > org.apache.zeppelin.scheduler.ParallelScheduler.lambda$runJobInScheduler$0(ParallelScheduler.java:39) > at > > java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) > at > > java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) > at java.lang.Thread.run(Thread.java:748) > Caused by: org.apache.zeppelin.interpreter.InterpreterException: > java.lang.NoSuchMethodError: > > org.apache.flink.api.common.ExecutionConfig.disableSysoutLogging()Lorg/apache/flink/api/common/ExecutionConfig; > at > > org.apache.zeppelin.interpreter.LazyOpenInterpreter.open(LazyOpenInterpreter.java:76) > at > > org.apache.zeppelin.interpreter.Interpreter.getInterpreterInTheSameSessionByClassName(Interpreter.java:355) > at > > org.apache.zeppelin.interpreter.Interpreter.getInterpreterInTheSameSessionByClassName(Interpreter.java:366) > at > > org.apache.zeppelin.flink.FlinkStreamSqlInterpreter.open(FlinkStreamSqlInterpreter.java:47) > at > > org.apache.zeppelin.interpreter.LazyOpenInterpreter.open(LazyOpenInterpreter.java:70) > ... 8 more > Caused by: java.lang.NoSuchMethodError: > > org.apache.flink.api.common.ExecutionConfig.disableSysoutLogging()Lorg/apache/flink/api/common/ExecutionConfig; > at > > org.apache.zeppelin.flink.FlinkScalaInterpreter.setTableEnvConfig(FlinkScalaInterpreter.scala:444) > at > > org.apache.zeppelin.flink.FlinkScalaInterpreter.open(FlinkScalaInterpreter.scala:114) > at > org.apache.zeppelin.flink.FlinkInterpreter.open(FlinkInterpreter.java:67) > at > > org.apache.zeppelin.interpreter.LazyOpenInterpreter.open(LazyOpenInterpreter.java:70) > ... 12 more > -- Best Regards Jeff Zhang
Re: flink zeppelin的type参数(append/update/single)和flink的动态表有关系嘛
和 flink 没关系，是zeppelin自己定义的参数，只影响select 语句，对于zeppelin的数据可视化有影响，不影响flink job 赵一旦 于2020年11月23日周一 下午1:11写道： > 如题，这个参数是仅仅zeppelin自身的参数，用于决定如何展示数据之类的逻辑呢？ > 还是和flink任务也有关系。按照 > https://www.yuque.com/jeffzhangjianfeng/gldg8w/te2l1c#AzSOu > 说法，append模式第一个结果列必须是时间，所以看起来更像是zeppelin自身的要求。 > > 我看了下append方式执行，jdbc仍然使用的upsertSink。 > > 所以谁确认下这个参数是不是和具体任务没啥关系。 > -- Best Regards Jeff Zhang
Re: Flink作业运行失败
你是hadoop2 吗？我记得这个情况只有hadoop3才会出现 gangzi <1139872...@qq.com> 于2020年10月16日周五 上午11:22写道： > TM > 的CLASSPATH确实没有hadoop-mapreduce-client-core.jar。这个难道是hadoop集群的问题吗？还是一定要shade-hadoop包，官方不推荐shade-hadoop包了。 > > > 2020年10月16日 上午10:50，Jeff Zhang 写道： > > > > 你看看TM的log，里面有CLASSPATH的 > > > > gangzi <1139872...@qq.com> 于2020年10月16日周五 上午10:11写道： > > > >> 我按照flink官方文档的做法，在hadoop集群每个节点上都：export HADOOP_CLASSPATH =`hadoop > >> classpath`,但是报：java.lang.NoClassDefFoundError: > >> org/apache/hadoop/mapred/JobConf > >> > >> > 不知道这个是不是flink的bug,按照这个报错，是缺少：hadoop-mapreduce-client-core.jar这个jar包，但是这个包是在/usr/local/hadoop-2.10.0/share/hadoop/mapreduce/*:这个目录下的，这个目录是包含在HADOOP_CLASSPATH上的，按理说是会加载到的。 > >> > >>> 2020年10月16日 上午9:59，Shubin Ruan 写道： > >>> > >>> export HADOOP_CLASSPATH= > >> > >> > > > > -- > > Best Regards > > > > Jeff Zhang > > -- Best Regards Jeff Zhang
Re: Flink作业运行失败
你看看TM的log，里面有CLASSPATH的 gangzi <1139872...@qq.com> 于2020年10月16日周五 上午10:11写道： > 我按照flink官方文档的做法，在hadoop集群每个节点上都：export HADOOP_CLASSPATH =`hadoop > classpath`,但是报：java.lang.NoClassDefFoundError: > org/apache/hadoop/mapred/JobConf > > 不知道这个是不是flink的bug,按照这个报错，是缺少：hadoop-mapreduce-client-core.jar这个jar包，但是这个包是在/usr/local/hadoop-2.10.0/share/hadoop/mapreduce/*:这个目录下的，这个目录是包含在HADOOP_CLASSPATH上的，按理说是会加载到的。 > > > 2020年10月16日 上午9:59，Shubin Ruan 写道： > > > > export HADOOP_CLASSPATH=**** > > -- Best Regards Jeff Zhang
Re: 动态加载table和udf的方法
zeppelin 支持直接写udf，参考 https://www.yuque.com/jeffzhangjianfeng/gldg8w/dthfu2 或者加入钉钉群讨论：32803524 Zeahoo Z 于2020年10月10日周六 上午9:04写道： > 你好，在开发中遇到了下面这个困难。 > > 目前将定义的table和udf函数写在了 conf/sql-client-defaults.yaml > > 文件中。程序运行没有问题，但是带来一个问题：如果我需要添加或者修改table或者udf文件的时候，需要重启flink程序。有没有办法可以让我动态地添加或者更新我的table和udf（table新增可以通过sql-client来添加，但是更倾向于将table的定义记录在文件，udf存在修改和新增的情况）。这样依赖可以保证flink不重启。 > -- Best Regards Jeff Zhang
Re: 怎么样在Flink中使用java代码提交job到yarn
如果不是flink jar的job，可以用zeppelin sdk来提交flink job https://www.yuque.com/jeffzhangjianfeng/gldg8w/pz2xoh xiao cai 于2020年9月25日周五 下午4:23写道： > Hi all： > 大家好，我目前遇到一个flink 任务提交方面的困扰： > 想要在自己的项目中（springboot）提交flink > job到yarn上。目前采用的方式是process类在本地起进程，调用bin/flink来启动。这样不是很优雅，而且比较难获取到提交的jobId和containerId。看到网上有博客使用ClusterClient的方式来提交，但是遇到了classpath的问题，会缺失一些FLINK_HOME/lib下的jar包 > 所以我想问，应该如何在自己的应用中提交任务到yarn，既能拿到任务信息，又可以解决classpath为指定带来的困扰。 > 非常感谢 > > > best, > xiao -- Best Regards Jeff Zhang
Re: zeppelin指定的依赖去哪找
没有搜索路径，需要用绝对路径 赵一旦 于2020年9月24日周四 下午3:22写道： > 看了你文章，有jars。想继续问下，jars必须完全路径嘛，有没有什么默认的搜索路径，我简单写jar名字的。不想写死路径。 > > 赵一旦 于2020年9月24日周四 下午3:17写道： > > > 这就有点麻烦了，公司机器一般不允许连接外部网络的。 > > > > Jeff Zhang 于2020年9月24日周四 下午3:15写道： > > > >> flink.execution.packages 下的包会从maven仓库下载，看这里关于如何在zeppelin里管理第三方依赖的方法 > >> https://www.yuque.com/jeffzhangjianfeng/gldg8w/rn6g1s > >> > >> > >> 赵一旦 于2020年9月24日周四 下午3:09写道： > >> > >> > 通过zeppelin写sql，之前sql-client可行的sql总是报错。 > >> > > >> > zeppelin配置的FLINK_HOME的目录中lib下是有相关包的，但远程remote集群没有。 > >> > 之前sql-client方式是基于-l方式指定的。 > >> > > >> > > >> > zeppelin情况下，貌似有个flink.execution.packages，但是并没说明这个指定的包去哪找的？是zeppelin配置的FLINK_HOME中的lib嘛？我lib中有包，但还是报错。 > >> > > >> > >> > >> -- > >> Best Regards > >> > >> Jeff Zhang > >> > > > -- Best Regards Jeff Zhang
Re: zeppelin指定的依赖去哪找
两个办法： 1. 用私有的maven 仓库 2. 自己打jar包，用 flink.exection.jars 来指定 https://www.yuque.com/jeffzhangjianfeng/gldg8w/rn6g1s#3BNYl 赵一旦 于2020年9月24日周四 下午3:17写道： > 这就有点麻烦了，公司机器一般不允许连接外部网络的。 > > Jeff Zhang 于2020年9月24日周四 下午3:15写道： > > > flink.execution.packages 下的包会从maven仓库下载，看这里关于如何在zeppelin里管理第三方依赖的方法 > > https://www.yuque.com/jeffzhangjianfeng/gldg8w/rn6g1s > > > > > > 赵一旦 于2020年9月24日周四 下午3:09写道： > > > > > 通过zeppelin写sql，之前sql-client可行的sql总是报错。 > > > > > > zeppelin配置的FLINK_HOME的目录中lib下是有相关包的，但远程remote集群没有。 > > > 之前sql-client方式是基于-l方式指定的。 > > > > > > > > > zeppelin情况下，貌似有个flink.execution.packages，但是并没说明这个指定的包去哪找的？是zeppelin配置的FLINK_HOME中的lib嘛？我lib中有包，但还是报错。 > > > > > > > > > -- > > Best Regards > > > > Jeff Zhang > > > -- Best Regards Jeff Zhang
Re: zeppelin指定的依赖去哪找
flink.execution.packages 下的包会从maven仓库下载，看这里关于如何在zeppelin里管理第三方依赖的方法 https://www.yuque.com/jeffzhangjianfeng/gldg8w/rn6g1s 赵一旦 于2020年9月24日周四 下午3:09写道： > 通过zeppelin写sql，之前sql-client可行的sql总是报错。 > > zeppelin配置的FLINK_HOME的目录中lib下是有相关包的，但远程remote集群没有。 > 之前sql-client方式是基于-l方式指定的。 > > zeppelin情况下，貌似有个flink.execution.packages，但是并没说明这个指定的包去哪找的？是zeppelin配置的FLINK_HOME中的lib嘛？我lib中有包，但还是报错。 > -- Best Regards Jeff Zhang
Re: 多线程模式下使用Blink TableEnvironment
Hi jun su, 如果是自建平台的话，可以考虑用zeppelin的sdk 来提交作业 https://www.yuque.com/jeffzhangjianfeng/gldg8w/pz2xoh jun su 于2020年9月18日周五 上午10:59写道： > hi godfrey, > > 我们的用法类似zeppelin, 项目形式类似notebook, 在第一次运行笔记时创建env, > 再次运行notebook时会创建新线程来构建job运行, 所以我参考了zepplin的做法暂时fix了这个问题 > > godfrey he 于2020年9月17日周四 下午10:07写道： > > > TableEnvironment 不是多线程安全的。 > > > > btw, 你能描述一下你在多线程情况下怎么使用 TableEnvironment 的吗？ > > > > Jeff Zhang 于2020年9月14日周一 下午12:10写道： > > > > > 参考zeppelin的做法，每个线程里都调用这个 > > > > > > > > > > > > https://github.com/apache/zeppelin/blob/master/flink/interpreter/src/main/java/org/apache/zeppelin/flink/FlinkSqlInterrpeter.java#L111 > > > > > > > > > jun su 于2020年9月14日周一 上午11:54写道： > > > > > > > hi all, > > > > > > > > 多线程模式下执行sql ， 在非聚合sql时报了如下错误： > > > > > > > > Caused by: java.lang.NullPointerException > > > > at java.util.Objects.requireNonNull(Objects.java:203) > > > > at > > > > > > > > > > > > > > org.apache.calcite.rel.metadata.RelMetadataQuery.(RelMetadataQuery.java:141) > > > > at > > > > > > > > > > > > > > org.apache.calcite.rel.metadata.RelMetadataQuery.(RelMetadataQuery.java:106) > > > > > > > > > > > > > > > > > > > > > > 已经用RelMetadataQueryBase.THREAD_PROVIDERS.set(JaninoRelMetadataProvider.of(FlinkDefaultRelMetadataProvider.INSTANCE())) > > > > 解决 > > > > > > > > > > > > 但是执行聚合sql时 ， 仍然会报错， 请问有办法临时fix？ > > > > > > > > Caused by: java.lang.NullPointerException > > > > at scala.Predef$.Double2double(Predef.scala:365) > > > > at > > > > > > > > > > > > > > org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalAggregate.computeSelfCost(FlinkLogicalAggregate.scala:81) > > > > at > > > > > > > > > > > > > > org.apache.calcite.rel.metadata.RelMdPercentageOriginalRows.getNonCumulativeCost(RelMdPercentageOriginalRows.java:174) > > > > at > > > > > > GeneratedMetadataHandler_NonCumulativeCost.getNonCumulativeCost_$(Unknown > > > > Source) > > > > at > > > GeneratedMetadataHandler_NonCumulativeCost.getNonCumulativeCost(Unknown > > > > Source) > > > > > > > > -- > > > > Best, > > > > Jun Su > > > > > > > > > > > > > -- > > > Best Regards > > > > > > Jeff Zhang > > > > > > > > -- > Best, > Jun Su > -- Best Regards Jeff Zhang
Re: 多线程模式下使用Blink TableEnvironment
参考zeppelin的做法，每个线程里都调用这个 https://github.com/apache/zeppelin/blob/master/flink/interpreter/src/main/java/org/apache/zeppelin/flink/FlinkSqlInterrpeter.java#L111 jun su 于2020年9月14日周一 上午11:54写道： > hi all, > > 多线程模式下执行sql ， 在非聚合sql时报了如下错误： > > Caused by: java.lang.NullPointerException > at java.util.Objects.requireNonNull(Objects.java:203) > at > > org.apache.calcite.rel.metadata.RelMetadataQuery.(RelMetadataQuery.java:141) > at > > org.apache.calcite.rel.metadata.RelMetadataQuery.(RelMetadataQuery.java:106) > > > > 已经用RelMetadataQueryBase.THREAD_PROVIDERS.set(JaninoRelMetadataProvider.of(FlinkDefaultRelMetadataProvider.INSTANCE())) > 解决 > > > 但是执行聚合sql时 ， 仍然会报错， 请问有办法临时fix？ > > Caused by: java.lang.NullPointerException > at scala.Predef$.Double2double(Predef.scala:365) > at > > org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalAggregate.computeSelfCost(FlinkLogicalAggregate.scala:81) > at > > org.apache.calcite.rel.metadata.RelMdPercentageOriginalRows.getNonCumulativeCost(RelMdPercentageOriginalRows.java:174) > at > GeneratedMetadataHandler_NonCumulativeCost.getNonCumulativeCost_$(Unknown > Source) > at GeneratedMetadataHandler_NonCumulativeCost.getNonCumulativeCost(Unknown > Source) > > -- > Best, > Jun Su > -- Best Regards Jeff Zhang
Re: 有木有比较好flink sql 任务提交插件推荐
zeppelin 提交sql 也很方便，可以通过UI，也可以通过zeppelin sdk， https://www.yuque.com/jeffzhangjianfeng/gldg8w/pz2xoh 钉钉群号：32803524 二维码： [image: flink_on_zeppelin_2.png] xuzh 于2020年9月14日周一 上午10:01写道： > Dear all: > 目前有找到两个sql任务提交插件： > https://github.com/wuchong/flink-sql-submit > https://github.com/springMoon/sqlSubmit > 大家有木有用过，推荐一下 -- Best Regards Jeff Zhang
Re: flink集成spring
可以看看这个zeppelin sdk,，也许适合你 https://www.yuque.com/jeffzhangjianfeng/gldg8w/pz2xoh 1115098...@qq.com 于2020年9月10日周四 上午9:09写道： > 大家好，我在将spring boot集成到flink的过程中，遇到很多问题，感觉不太兼容。看官方文档也没有集成spring > boot的介绍，是不是flink设计的时候就没有考虑与spring boot的集成？ -- Best Regards Jeff Zhang
Re: flink sql client 如何同时执行多条 sql 语句
可以尝试下zeppelin，zeppelin是支持多条sql的。可以看这里的文档和加入钉钉群 https://www.yuque.com/jeffzhangjianfeng/gldg8w/te2l1c#WMEMY 钉钉群号：32803524 钉钉群二维码： [image: flink_on_zeppelin_2.png] faaron zheng 于2020年9月5日周六 上午9:37写道： > Hi, sql-client目前应该是没有这个能力的，它是交互式执行的，我们之前在sql-client的基础上改过一个类似beeline > -e/-f的脚本，主要修改的提交任务的地方。 Best， Faaron Zheng 在2020年09月04日 17:04，LittleFall 写道： > 我有一个 sql 文件，它里面有不少 flink sql 的创建表的语句和查询语句，现在我想要通过 sql client > 提交这些任务，却只能一句一句的复制粘贴。如果同时复制粘贴多条语句就会报错，在 flink sql client 中使用 source xxx.sql > 也会报错。 请问用什么样的方法可以一次性执行多条语句呢？ -- Sent from: > http://apache-flink.147419.n8.nabble.com/ -- Best Regards Jeff Zhang
Re: 基于flink1.10源码编译问题
不要用aliyun maven repo，另外你这是1.10-SNAPSHOT 不是1.10的release版本 魏烽 于2020年8月21日周五 下午8:44写道： > 各位好： > > >
Re: flink任务提交
Zeppelin 最近在做这样的API来提交Flink Job，这里有个例子可以参考下 https://github.com/zjffdu/zeppelin/blob/ZEPPELIN-4981/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/ZSessionIntegrationTest.java#L307 可以加入钉钉群讨论，钉钉群号：32803524 Dream-底限 于2020年8月19日周三 下午4:27写道： > hi、 > 请问你说的是自己拼接cli字符串，然后通过java调用子进程执行的这种方式吗 > > 我先前也是这么做的，但感觉比较怪异，这种方式有一个问题是貌似没办法直接返回applicationId，要从日志里面筛选好像，再就是没办法判断提交是否成功，貌似也是从日志来做，请问这个applicationId在提交的时候除了从日志筛选有其他的获取方式吗 > > wxpcc 于2020年8月19日周三 下午4:09写道： > > > 大概可以用，YarnClusterDescriptor > > > > 命令行方式封装提交对于后续升级更加方便一些，个人建议 > > > > > > > > -- > > Sent from: http://apache-flink.147419.n8.nabble.com/ > -- Best Regards Jeff Zhang
Re: FlinkSQL命令行方式提交sql任务的jar依赖问题
你的10台机器是flink standalone还是 yarn集群 ？ 如果是yarn集群，那么只要把依赖放到机器A就可以了，如果是standalone的话，就要在B集群上部署了依赖了。 另外你可以用zeppelin来提交flink sql 作业，zeppelin提供很多种添加第三方依赖的方法，具体可以参考文档 https://www.yuque.com/jeffzhangjianfeng/gldg8w/rn6g1s 或者加入钉钉群讨论，钉钉群号: 32803524 Zhao,Yi(SEC) 于2020年8月13日周四 下午1:02写道： > 背景，我的flink集群（10机器）A，此外还有一个单独的flink机器（作为提交任务的机器）B。 > 现在比较混乱，哪些jar需要放到A，哪些放到B。 > > > （1） kafka ssl > 证书，这个肯定需要放到A，因为kafka提供的属性只是ssl证书的路径，真正读取证书是任务开始运行之后，因此这个路径必须是集群A可访问，B是否可访问其实无所谓。 > > （2） > > flink-sql情况下的kafak-client包。这个比较奇怪，从我的实验效果来看，貌似机器B上没有会报错，必须在启动sql-client的时候通过-j或-l参数指定jar路径，并包含kafka-client.jar,类似的还有flink-json等格式包。此外，这些包在集群A上都没有，但运行都正常。这么来看，我在sql-client中提交一个sql的时候，部分依赖应该是被打包并且上传到集群A的？这是怎么决定的呢？从我的sql中动态判定哪些包用到？还是咋的。 > > > > 总结下，FlinkSQL使用sql-client命令行方式提交sql执行任务的场景下。最终运行时候哪些包需要用到sql-client端的包呢？ > > 目前从实验来看Flink-json、各种connector的包都是用的sql-client的包。 > > > -- Best Regards Jeff Zhang
Re: 数据预览
Apache Zeppelin有自己的rest api，你可以用rest api来提交flink sql 以及拿sql结果，目前Zeppelin社区正在做一个Client API （Zeppelin SDK), 用户可以更加方便的调用Zeppelin的功能。具体可以参考 https://issues.apache.org/jira/browse/ZEPPELIN-4981 这里有Sample code 可以参考 https://github.com/zjffdu/zeppelin/blob/ZEPPELIN-4981/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/ZSessionIntegrationTest.java#L298 对于Flink on Zeppelin感兴趣的，可以加入钉钉群：32803524 forideal 于2020年8月1日周六 下午7:49写道： > 你好，我的朋友 > > >最近我看 Flink doc 中的文档中有了如下 connector > DataGen > Print > BlackHole >这大大的方便了开发和调试。不过，我还是不太满足，想了解一下数据预览相关的做法。 >比如我想，如果我有一个 Flink 的 `driver` ，然后，我使用这个 driver 提交一条 SQL，我从 ResultSet > 中获取数据。这样又可以大大的方面我们的 Flink SQL 开发者。 >在社区中，我已经体验了 Apache Zeppelin ，他可以让我提交 Flink SQL，然后在页面上面等待刷新的结果，但是 > Zeppelin 目前不能很好的集成到我们的 Flink web IDE 中。想了解一下如何实现数据预览。 > > >Best forideal -- Best Regards Jeff Zhang
Re: 关于 sql-client
可以用zeppelin来提交flink sql作业，可以加入钉钉群讨论：32803524
杨荣 于2020年7月24日周五 下午3:19写道：
> 你们可以 pr 到官方啊。我觉得这个功能很刚需啊，并且 basic 版的 1.5 就 release 了，不知道为什么相关 gate way 或者
> submit with sql file 的 feature 到现在都还没实现呢。
>
> Harold.Miao 于2020年7月24日周五 上午11:42写道：
>
> > 1 应该是可以的 主要是你要在flink-conf.yaml里面配置正确的 jobmanager.rpc.address
> > 源码里面有加载主配置文件的逻辑
> >
> > public LocalExecutor(URL defaultEnv, List jars, List
> libraries) {
> >// discover configuration
> >final String flinkConfigDir;
> >try {
> > // find the configuration directory
> > flinkConfigDir = CliFrontend.getConfigurationDirectoryFromEnv();
> >
> > // load the global configuration
> > this.flinkConfig =
> > GlobalConfiguration.loadConfiguration(flinkConfigDir);
> >
> > // initialize default file system
> > FileSystem.initialize(flinkConfig,
> > PluginUtils.createPluginManagerFromRootFolder(flinkConfig));
> >
> > // load command lines for deployment
> > this.commandLines =
> > CliFrontend.loadCustomCommandLines(flinkConfig, flinkConfigDir);
> > this.commandLineOptions = collectCommandLineOptions(commandLines);
> >} catch (Exception e) {
> > throw new SqlClientException("Could not load Flink configuration.",
> > e);
> >}
> >
> >
> > 2 因为等不及官方的 我们自己wrapper实现了一个
> >
> >
> >
> >
> > 杨荣 于2020年7月24日周五 上午10:53写道：
> >
> > > Hi all,
> > >
> > > 请问：
> > > 1. 在 Embedded mode 下，支持 ClusterClient 进行 job
> > > 提交作业，进行分布式计算吗？在文档中没看到，跟着文档走，只启起了 Local 在本地作业，无法运用到生产环境。
> > >
> > > 2. GateWay mode 预计在那个版本 release？
> > >
> >
> >
> > --
> >
> > Best Regards,
> > Harold Miao
> >
>
--
Best Regards
Jeff Zhang
Re: FlinkSQL 任务提交后 任务名称问题
在zeppelin中你可以指定insert 语句的job name，如下图，（对Zeppelin感兴趣的，可以加入钉钉群：32803524）
%flink.ssql(jobName="my job")
insert into sink_kafka select status, direction, cast(event_ts/10
as timestamp(3)) from source_kafka where status <> 'foo'
[image: image.png]
Evan 于2020年7月18日周六 下午5:47写道：
> 代码大概是这样子的，一张kafka source表，一张es Sink表，最后通过tableEnv.executeSql("insert into
> esSinkTable select ... from kafkaSourceTable")执行
> 任务提交后任务名称为“inset-into_某某catalog_某某database.某某Table”
>
>
> 这样很不友好啊，能不能我自己指定任务名称呢？
--
Best Regards
Jeff Zhang
Re: 代码中如何取消正在运行的Flink Streaming作业
Zeppelin 能够帮你提交和cancel job，就是通过上面jianxu说的ClusterClient api来做到的，对zeppelin感兴趣的话，可以参考这个视频 https://www.bilibili.com/video/BV1Te411W73b?p=21 jianxu 于2020年7月11日周六 下午4:52写道： > Hi: > > 我想，你可能打算通过API的方式来取消正在运行的流任务。Flink任务提交时需要构建ClusterClient，提交成功后会返回任务对应的JobId。任务取消时，通过调用ClusterClient的cancel(JobID > jobId)取消流任务。 > Flink源码可以看看 CliFrontend[1]中的逻辑，如果觉得比较麻烦可以参考 > https://github.com/todd5167/flink-spark-submiter > 项目的任务提交部分，取消任务时构建ClusterClient即可。 > > > > > > > > > > > | | > jianxu > | > | > rjia...@163.com > | > > > > > 在2020年07月11日 16:19，Congxian Qiu 写道： > Hi > > 如果你是想做一个作业管理的平台，可以尝试看一下 CliFrontend[1] 中相关的逻辑，对于 On Yarn > 的作业，简单地说你需要能够正确的初始化一个 client 和 Yarn RM 交互，然后你需要知道 applicationId，另外你还需要知道 > flink 的 JobId，接下来就是调用 Flink 的接口了 > > 如果像更多的了解参数如从和命令行传到 java 代码的，你可以自己写一个单元测试，单步调试一下整个流程。 > > [1] > > https://github.com/apache/flink/blob/master/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java > > Best, > Congxian > > > godfrey he 于2020年7月9日周四 上午10:08写道： > > 可以通过 StreamExecutionEnvironment#executeAsync 提交作业，返回 JobClient [1], 通过 > JobClient 可以 cancel 作业，获取 job status。 > > [1] > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-74%3A+Flink+JobClient+API > > Best, > Godfrey > > Evan 于2020年7月9日周四 上午9:40写道： > > 这个问题之前看到过有人在问，但是没有看到答案，我想问一下，Flink Streaming > API有没有提供类似的接口，调用后就能停止这个Stream作业呢？ > > -- Best Regards Jeff Zhang
Re: pyflink数据查询
可以用zeppelin的z.show 来查询job结果。这边有pyflink在zeppelin上的入门教程 https://www.bilibili.com/video/BV1Te411W73b?p=20 可以加入钉钉群讨论：30022475 jack 于2020年6月9日周二 下午5:28写道： > 问题请教： > 描述： pyflink 从source通过sql对数据进行查询聚合等操作 > 不输出到sink中，而是可以直接作为结果，我这边可以通过开发web接口直接查询这个结果，不必去sink中进行查询。 > > flink能否实现这样的方式？ > 感谢 > -- Best Regards Jeff Zhang
Re: 關於LocalExecutionEnvironment使用MiniCluster的配置
在zeppelin也集成了flink的local 模式，可以通过设置 local.number-taskmanager 和 flink.tm.slot来设置tm和slot的数目， 具体可以参考这个视频 https://www.bilibili.com/video/BV1Te411W73b?p=3 tison 于2020年5月24日周日 下午9:46写道： > 是这样的。 > > 这里的配置可以参考[1][2]两个类，具体你 Maven 启动的代码路径还跟[3][4]有关。 > > 这边可能确实文档比较缺失。可以看下配置传递的路径，TM 的数量还有 RPC 的共享格式等配置，至少编程接口上都是可以配的。 > > Best, > tison. > > [1] > > https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java > [2] > > https://github.com/apache/flink/blob/ab947386ed93b16019f36c50e9a3475dd6ad3c4a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterConfiguration.java > [3] > > https://github.com/apache/flink/blob/master/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/LocalExecutor.java > [4] > > https://github.com/apache/flink/blob/master/flink-clients/src/main/java/org/apache/flink/client/program/PerJobMiniClusterFactory.java > > > > > 月月 于2020年5月24日周日 下午9:11写道： > > > 您好， > > 在單機模式使用maven執行專案時，會自動啟動MiniCluster， > > 我想請問在這種情形下，預設是配置一個JobManager以及一個TaskManager嗎? > > > > 找了一下文件中並沒有相關的說明。 > > > > 感謝！ > > > -- Best Regards Jeff Zhang
Re: 按照官网进行flink-shell操作，出现无法解决的错误：Only BatchTableSource and InputFormatTableSource are supported in BatchTableEnvironment.
可以在zeppelin里写scala代码，是支持hive的，参考这个视频，
https://www.bilibili.com/video/BV1Te411W73b?p=10
也可以到这个钉钉群讨论: 30022475
Jingsong Li 于2020年5月21日周四 下午4:43写道：
> Hi,
>
> 不好意思，现在版本hive connector已经不支持old planner了，
> 但是scala shell还是默认old planner。
>
> Best,
> Jingsong Lee
>
> On Thu, May 21, 2020 at 3:24 PM 张锴 wrote:
>
> > 具体操作及错误信息我贴到下面，各位大佬帮忙看下如何解决，不知道是不是BUG。
> >
> > scala> import org.apache.flink.table.catalog.hive.HiveCatalog
> > import org.apache.flink.table.catalog.hive.HiveCatalog
> >
> > scala> val hiveCatalog = new HiveCatalog("hive", "mydatabase",
> > "/opt/hive2.3.3/conf", "2.3.3");
> > hiveCatalog: org.apache.flink.table.catalog.hive.HiveCatalog =
> > org.apache.flink.table.catalog.hive.HiveCatalog@193f3306
> >
> > scala> btenv.registerCatalog("hive", hiveCatalog)
> > Thu May 21 15:10:05 CST 2020 WARN: Establishing SSL connection without
> > server's identity verification is not recommended. According to MySQL
> > 5.5.45+, 5.6.26+ and 5.7.6+ requirements SSL connection must be
> established
> > by default if explicit option isn't set. For compliance with existing
> > applications not using SSL the verifyServerCertificate property is set to
> > 'false'. You need either to explicitly disable SSL by setting
> useSSL=false,
> > or set useSSL=true and provide truststore for server certificate
> > verification.
> > Thu May 21 15:10:05 CST 2020 WARN: Establishing SSL connection without
> > server's identity verification is not recommended. According to MySQL
> > 5.5.45+, 5.6.26+ and 5.7.6+ requirements SSL connection must be
> established
> > by default if explicit option isn't set. For compliance with existing
> > applications not using SSL the verifyServerCertificate property is set to
> > 'false'. You need either to explicitly disable SSL by setting
> useSSL=false,
> > or set useSSL=true and provide truststore for server certificate
> > verification.
> >
> > scala> btenv.useCatalog("hive")
> >
> > scala> btenv.listTables
> > res2: Array[String] = Array(mytable)
> >
> > scala> btenv.sqlQuery("select * from mytable").toDataSet[Row].print()
> > org.apache.flink.table.api.TableException: Only BatchTableSource and
> > InputFormatTableSource are supported in BatchTableEnvironment.
> > at
> >
> >
> org.apache.flink.table.plan.nodes.dataset.BatchTableSourceScan.translateToPlan(BatchTableSourceScan.scala:118)
> > at
> >
> >
> org.apache.flink.table.api.internal.BatchTableEnvImpl.translate(BatchTableEnvImpl.scala:306)
> > at
> >
> >
> org.apache.flink.table.api.internal.BatchTableEnvImpl.translate(BatchTableEnvImpl.scala:281)
> > at
> >
> >
> org.apache.flink.table.api.scala.internal.BatchTableEnvironmentImpl.toDataSet(BatchTableEnvironmentImpl.scala:69)
> > at
> >
> >
> org.apache.flink.table.api.scala.TableConversions.toDataSet(TableConversions.scala:53)
> > ... 30 elided
> >
>
>
> --
> Best, Jingsong Lee
>
--
Best Regards
Jeff Zhang
Re: spark代码直接运行至Flink平台
啥目的 ？ hsdcl...@163.com 于2020年4月22日周三 上午9:49写道： > Hi, > 有个脑洞大开的想法，有没有可能出一个模块功能，可以将用户写的spark代码直接运行在Flink平台 -- Best Regards Jeff Zhang
Re: [Third-party Tool] Flink memory calculator
Hi Yangze, Does this tool just parse the configuration in flink-conf.yaml ? Maybe it could be done in JobListener [1] (we should enhance it via adding hook before job submission), so that it could all the cases (e.g. parameters coming from command line) [1] https://github.com/apache/flink/blob/master/flink-core/src/main/java/org/apache/flink/core/execution/JobListener.java#L35 Yangze Guo 于2020年3月30日周一 上午9:40写道： > Hi, Yun, > > I'm sorry that it currently could not handle it. But I think it is a > really good idea and that feature would be added to the next version. > > Best, > Yangze Guo > > On Mon, Mar 30, 2020 at 12:21 AM Yun Tang wrote: > > > > Very interesting and convenient tool, just a quick question: could this > tool also handle deployment cluster commands like "-tm" mixed with > configuration in `flink-conf.yaml` ? > > > > Best > > Yun Tang > > > > From: Yangze Guo > > Sent: Friday, March 27, 2020 18:00 > > To: user ; user-zh@flink.apache.org < > user-zh@flink.apache.org> > > Subject: [Third-party Tool] Flink memory calculator > > > > Hi, there. > > > > In release-1.10, the memory setup of task managers has changed a lot. > > I would like to provide here a third-party tool to simulate and get > > the calculation result of Flink's memory configuration. > > > > Although there is already a detailed setup guide[1] and migration > > guide[2] officially, the calculator could further allow users to: > > - Verify if there is any conflict in their configuration. The > > calculator is more lightweight than starting a Flink cluster, > > especially when running Flink on Yarn/Kubernetes. User could make sure > > their configuration is correct locally before deploying it to external > > resource managers. > > - Get all of the memory configurations before deploying. User may set > > taskmanager.memory.task.heap.size and taskmanager.memory.managed.size. > > But they also want to know the total memory consumption of Flink. With > > this tool, users could get all of the memory configurations they are > > interested in. If anything is unexpected, they would not need to > > re-deploy a Flink cluster. > > > > The repo link of this tool is > > https://github.com/KarmaGYZ/flink-memory-calculator. It reuses the > > BashJavaUtils.jar of Flink and ensures the calculation result is > > exactly the same as your Flink dist. For more details, please take a > > look at the README. > > > > Any feedback or suggestion is welcomed! > > > > [1] > https://ci.apache.org/projects/flink/flink-docs-master/ops/memory/mem_setup.html > > [2] > https://ci.apache.org/projects/flink/flink-docs-master/ops/memory/mem_migration.html > > > > Best, > > Yangze Guo > -- Best Regards Jeff Zhang
Re: 关于sql-client和sql-gateway sql中的注释支持问题
Hi xinghalo， Zeppelin有rest api，可以用调度系统来调rest api来跑Zeppelin的flink sql，不一定要通过UI的 111 于2020年3月13日周五 下午5:29写道： > Hi Jeff， > 好滴，已经加入交流群了，多交流。 > 不过我们目前是基于flink做的数据分析平台，可以直接集成很多任务调度的功能。 > 目前手动去除了注释，再提交，还是很期待官方自动支持这个特性的。 > > > | | > xinghalo > | > | > xingh...@163.com > | > 签名由网易邮箱大师定制 > > > 在2020年03月13日 16:34，Jeff Zhang 写道： > Hi xinghalo， > > 在Apache Zeppelin里运行Sql是可以支持sql comment的，可以加入钉钉群 30022475 来体验 > > godfrey he 于2020年3月13日周五 下午3:49写道： > > hi sql-gateway的做法目前和sql client类似，都是通过正则表达式来处理的。这一块sql-gateway的解法应该会和sql > client类似。我最近也会先在sql-gateway改用calcite来支持 > > Best, > Godfrey > > > Jingsong Li 于2020年3月13日周五 下午3:41写道： > > Hi, > > 现在SQL-CLI是自己解析的，等到后面统一到calcite来解析后，注释应该很自然的就支持了。 > @godfrey he 有计划统一到calcite来解析吗？ > > Best, > Jingsong Lee > > On Fri, Mar 13, 2020 at 3:23 PM 111 wrote: > > 您好： > > > 最近在使用sql-client和sql-gateway时，发现不支持注释。 > > > > Flink SQL> show tables; -- testtest ; > > [ERROR] Unknown or invalid SQL statement. > > > > > > 目前是每次使用sql-gateway提交时手动过滤掉注释，建议增加这种注释处理。 > > > | | > xinghalo > | > | > xingh...@163.com > | > 签名由网易邮箱大师定制 > > > > -- > Best, Jingsong Lee > > > > > -- > Best Regards > > Jeff Zhang > -- Best Regards Jeff Zhang
Re: 关于sql-client和sql-gateway sql中的注释支持问题
Hi xinghalo， 在Apache Zeppelin里运行Sql是可以支持sql comment的，可以加入钉钉群 30022475 来体验 godfrey he 于2020年3月13日周五 下午3:49写道： > hi sql-gateway的做法目前和sql client类似，都是通过正则表达式来处理的。这一块sql-gateway的解法应该会和sql > client类似。我最近也会先在sql-gateway改用calcite来支持 > > Best, > Godfrey > > > Jingsong Li 于2020年3月13日周五 下午3:41写道： > > > Hi, > > > > 现在SQL-CLI是自己解析的，等到后面统一到calcite来解析后，注释应该很自然的就支持了。 > > @godfrey he 有计划统一到calcite来解析吗？ > > > > Best, > > Jingsong Lee > > > > On Fri, Mar 13, 2020 at 3:23 PM 111 wrote: > > > >> 您好： > >> > >> > >> 最近在使用sql-client和sql-gateway时，发现不支持注释。 > >> > >> > >> > >> Flink SQL> show tables; -- testtest ; > >> > >> [ERROR] Unknown or invalid SQL statement. > >> > >> > >> > >> > >> > >> 目前是每次使用sql-gateway提交时手动过滤掉注释，建议增加这种注释处理。 > >> > >> > >> | | > >> xinghalo > >> | > >> | > >> xingh...@163.com > >> | > >> 签名由网易邮箱大师定制 > >> > >> > > > > -- > > Best, Jingsong Lee > > > -- Best Regards Jeff Zhang
Re: [DISCUSS] Upload the Flink Python API 1.9.x to PyPI for user convenience.
+1 Xingbo Huang 于2020年2月4日周二 下午1:07写道： > Hi Jincheng, > > Thanks for driving this. > +1 for this proposal. > > Compared to building from source, downloading directly from PyPI will > greatly save the development cost of Python users. > > Best, > Xingbo > > > > Wei Zhong 于2020年2月4日周二 下午12:43写道： > >> Hi Jincheng, >> >> Thanks for bring up this discussion! >> >> +1 for this proposal. Building from source takes long time and requires a >> good network environment. Some users may not have such an environment. >> Uploading to PyPI will greatly improve the user experience. >> >> Best, >> Wei >> >> jincheng sun 于2020年2月4日周二 上午11:49写道： >> >>> Hi folks, >>> >>> I am very happy to receive some user inquiries about the use of Flink >>> Python API (PyFlink) recently. One of the more common questions is >>> whether >>> it is possible to install PyFlink without using source code build. The >>> most >>> convenient and natural way for users is to use `pip install >>> apache-flink`. >>> We originally planned to support the use of `pip install apache-flink` in >>> Flink 1.10, but the reason for this decision was that when Flink 1.9 was >>> released at August 22, 2019[1], Flink's PyPI account system was not >>> ready. >>> At present, our PyPI account is available at October 09, 2019 [2](Only >>> PMC >>> can access), So for the convenience of users I propose: >>> >>> - Publish the latest release version (1.9.2) of Flink 1.9 to PyPI. >>> - Update Flink 1.9 documentation to add support for `pip install`. >>> >>> As we all know, Flink 1.9.2 was just completed released at January 31, >>> 2020 >>> [3]. There is still at least 1 to 2 months before the release of 1.9.3, >>> so >>> my proposal is completely considered from the perspective of user >>> convenience. Although the proposed work is not large, we have not set a >>> precedent for independent release of the Flink Python API(PyFlink) in the >>> previous release process. I hereby initiate the current discussion and >>> look >>> forward to your feedback! >>> >>> Best, >>> Jincheng >>> >>> [1] >>> >>> https://lists.apache.org/thread.html/4a4d23c449f26b66bc58c71cc1a5c6079c79b5049c6c6744224c5f46%40%3Cdev.flink.apache.org%3E >>> [2] >>> >>> https://lists.apache.org/thread.html/8273a5e8834b788d8ae552a5e177b69e04e96c0446bb90979444deee%40%3Cprivate.flink.apache.org%3E >>> [3] >>> >>> https://lists.apache.org/thread.html/ra27644a4e111476b6041e8969def4322f47d5e0aae8da3ef30cd2926%40%3Cdev.flink.apache.org%3E >>> >> -- Best Regards Jeff Zhang
Re: flink异常恢复
上个checkpoint 王金海 于2019年8月27日周二 下午6:14写道： > 讨论下flink异常重启问题 > > > 从kafka消费数据，checkpoint周期在5分钟，如果在第6分钟，因为异常导致flink任务重启，flink是从上个checkpoint恢复呢？还是从异常时的offset恢复呢？ > > > > csbl...@163.com > Have a nice day ! > > -- Best Regards Jeff Zhang
Re: flink基于yarn的方式提交，log在web上看太卡了。能不能直接看log文件？
你是通过flink UI看log还是yarn ui 看log ？ 陈帅 于2019年8月27日周二 下午5:55写道： > flink基于yarn的方式提交，log在web上看太卡了。能不能直接看log文件？ > -- Best Regards Jeff Zhang
Re: flink1.7.2如何进行hdfs的kerberos认证
See
https://ci.apache.org/projects/flink/flink-docs-release-1.9/ops/config.html#kerberos-based-security
杨文生-java开发 于2019年8月27日周二 下午3:30写道：
>
>
>public static void main(String[] args) throws Exception {
> final ParameterTool parameterTool = ParameterTool
>
> .fromPropertiesFile(BizlogStreamWithEventTimeCleaner.class.getResourceAsStream(PROPERTIES_FILE_NAME))
>
> .mergeWith(ParameterTool.fromArgs(args));
> String resultTable = parameterTool.get(BIZLOG_RESULT_TABLENAME);
> //设置窗口大小
> Time windowSize =
> Time.milliseconds(parameterTool.getLong(BIZLOG_WINDOW_SIZE));
> //设置数据最大乱序时间
> Time maxOutOfOrder =
> Time.milliseconds(parameterTool.getLong(BIZLOG_WINDOW_MAXOUTOFORDE));
> hdfsAuthenticate(parameterTool);
> ...
> }
> private static void hdfsAuthenticate(ParameterTool parameterTool)
> throws IOException {
> String kerberosConfFile =
> BizlogStreamWithEventTimeCleaner.class.getClassLoader().getResource("krb5.conf").getPath();
>
> System.setProperty("java.security.krb5.conf", kerberosConfFile);
> Configuration conf = new Configuration();
> conf.set(HADOOP_SECURITY_AUTHENTICATION, "kerberos");
> //加载hadoop配置文件
> String principal = parameterTool.get("kerberos.princpal");
> String keytabName = parameterTool.get("keytab.name");
> String keytabPath =
> BizlogStreamWithEventTimeCleaner.class.getClassLoader().getResource(keytabName).getPath();
>
> UserGroupInformation.setConfiguration(conf);
> UserGroupInformation.loginUserFromKeytab(principal, keytabPath);
> }
>
>
> 代码如上，在idea中直接运行可以认证通过，但是打成jar包提交到集群后报错如下：
> Caused by: java.io.IOException: Login failure for biuri/
> bj142.-in.dom...@btest.com from keytab
> file:/data/realtime-flink.jar!/kerberos.keytab:
> javax.security.auth.login.LoginException: Unable to obtain password from
> user
> 这个是什么原因？或者应该如何进行正确的集群认证？
>
>
>
>
--
Best Regards
Jeff Zhang
Re: kafka流与hive表join问题
你是怎么join hive表的，能share你的代码吗？ like 于2019年8月27日周二 下午5:15写道： > 请问一下各位大佬，如何保证先加载完hive表，然后再与流join，我发现在hive还没有加载完就已经有join的结果出来，这样刚开始出来的结果是不准确的，还有一个问题是hive表加载完之后不会再做checkpoint？我目前使用的是1.7.1版本，看了1.9的维表join，blink文档说（必须加上FOR > SYSTEM_TIME AS OF > PROCTIME()，表示JOIN维表当前时刻所看到的每条数据），这也就意味着跟我现在一样，没有加载完hive就会join输出了？ > > > > -- Best Regards Jeff Zhang
Re: Flink基础功能Flink for zeppelin痛点
你好，谢谢你的反馈 >>> 发现只支持Flink scala API Blink on Flink 是支持SQL，包括batch sql和streaming sql，你看的可能是apache zeppelin网站上的flink支持，我们对blink的zeppelin支持做了更多的工作，但是还没有merge到apache zeppelin中。具体信息你可以下面的页面， https://github.com/apache/flink/blob/blink/docs/quickstart/zeppelin_quickstart.md https://github.com/apache/flink/blob/blink/docs/ops/zeppelin.md >>> 希望社区对于Flink SQL客户端开发对应的JDBC 这个在我们的roadmap中 On 2019/02/01 07:26:25, "hanfeizi0409" wrote: > Hi,Everyone:> > >各位社区的小伙伴们，大家好！我是Flink社区的一名志愿者。在使用Flink for zeppelin时发现了2个痛点：> > > 首先，zeppelin for Flink在使用时，发现只支持Flink scala API。随着Blink 的开源，高级SQL的引入，Flink > SQL客户端> > 的开发也势在必得。比如在zeppelin中，用户不仅仅是scala去实现，可能有些用户只会SQL，对scala或者java根本不懂。> > 未来的趋势还是会回归SQL本质，另外对于spark sql用户也能很快切换过来。> > >第二点，希望社区对于Flink SQL客户端开发对应的JDBC，比如在dbeaver上通过JDBC连接Flink SQL;> > > 以上的建议是个人在使用上的一些痛点，希望对Flink 社区的发展有所帮助。谢谢~> > > > > >
