你的代码：w_ts as TO_TIMESTAMP(FROM_UNIXTIME(rowtime/1000),'-MM-dd HH:mm:ss')
这里里面的 FROM_UNIXTIME 函数就是接受BIGINT参数，返回 VARCHAR类型的日期值，默认日期格式：-MM-dd
HH:mm:ss
然后通过TO_TIMESTAMP函数转换成TIMESTAMP类型，是timestamp(3)类型，这个只能用blink planner
参考阿里云网址：https://help.aliyun.com/knowledge_list_page/6271
这个报错：>>> Caused by: org.apache.flink.runtime.JobException: Recovery is
suppressed
>>> by NoRestartBackoffTimeStrategy
应该是没有读取到flink conf下面的flink-conf.yaml配置文件，里面有个task失败重启配置参数！
发件人： air23
发送时间： 2020-05-29 14:34
收件人： user-zh
主题： Re:Re: flink1.10 on yarn 问题
代码就是flink自带的例子。
public class Word
你运行的命令是啥？然后在哪个目录下运行的，和 flink 下载下来解压的目录是什么相对关系？
Best,
tison.
air23 于2020年5月29日周五 下午2:35写道：
> 代码就是flink自带的例子。
>
> public class WordCountStreamingByJava {
> public static void main(String[] args) throws Exception {
>
> // 创建执行环境
> StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getE
代码就是flink自带的例子。
public class WordCountStreamingByJava {
public static void main(String[] args) throws Exception {
// 创建执行环境
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
// 设置socket数据源
DataStreamSource source = env.socketTextStream("zongteng75", 9001,
"\
请问，flink sql1.10有解析json的udf吗？没有找到
发自我的iPhone
你好 是cluster的 本地代码没有报错的 报错的消息贴下面了
flink1.7 时正常的。
后来我加上了flink的环境变量
#flink
export FLINK_HOME=/opt/module/flink-1.10.1
export PATH=${FLINK_HOME}/bin:$PATH
这个报错的例子 就正常跑了
但是换了另外一个任务 在1.7 和本地都是可以的。报错如下
The program finished with the follow
然后你 execute 前后的代码片段甚至整个 main 如果可以的话通过 gist 贴一下（x）
Best,
tison.
tison 于2020年5月29日周五 下午2:21写道：
> 这个问题好诡异啊，一般来说编译会在 env.execute
> 的时候拦截，不应该真的调度起来才对。你能详细描述一下你提交作业的方法还有这个错误报在哪里吗（client?cluster?）？
>
> Best,
> tison.
>
>
> air23 于2020年5月29日周五 下午1:38写道：
>
>> cdh运行flink1.10 on cdh yarn 报错如下。 用1.7.2版本就
这个问题好诡异啊，一般来说编译会在 env.execute
的时候拦截，不应该真的调度起来才对。你能详细描述一下你提交作业的方法还有这个错误报在哪里吗（client?cluster?）？
Best,
tison.
air23 于2020年5月29日周五 下午1:38写道：
> cdh运行flink1.10 on cdh yarn 报错如下。 用1.7.2版本就没有问题
> flink-shaded-hadoop-2-uber-2.6.5-10.0.jar 也加了
> hadoop环境变量 export HADOOP_CONF_DIR=/etc/hadoop/conf
> 求解答
>
目前在学习使用pyflink的Table api，请教一个问题：
1、Table Api连接kafka系统，能否把整条的kafka消息看成是一个table字段进行处理？比如，kafka
topic连的消息为一个json字符串，把这个字符串整体当做是一个字段，这样可以方便使用 pyflink 的udf函数对消息进行处理转换等操作？
2、如果以上可行，连接kafka的数据格式如何设置，即with_format如何设置，目前官网这方便的资料较少。
新手入门，请多指教，感谢。
cdh运行flink1.10 on cdh yarn 报错如下。 用1.7.2版本就没有问题
flink-shaded-hadoop-2-uber-2.6.5-10.0.jar 也加了
hadoop环境变量 export HADOOP_CONF_DIR=/etc/hadoop/conf
求解答
org.apache.flink.client.program.ProgramInvocationException: The main method
caused an error: org.apache.flink.client.program.ProgramInvocatio
可以看下你的HADOOP_CONF吗。我的配置的=/etc/hadoop/conf。
开源的Hadoop版本 这个也放了
在 2020-05-28 09:36:10，"wangweigu...@stevegame.cn"
写道：
>
>确实，你只要配置好CDH的HADOOP_CONF环境变量，同时下载开源的Hadoop版本（和CDH版本相同）放到flink
> lib下，就可以访问CDH yarn，提交作业！
>
>目前我这边是CDH 5.16.1，Flink 1.10，提交Flink on yarn是没问题，任务运行也没问题，
Hi,
> 我使用avro代码生成工具基于一个schema生成了UserView类，又实现了kafka的Serializer接口，
你往kafka中写入avro格式的数据时用kafak的Serializer，写入的格式是当成confluent schemat registry的avro格式吧
confluent schemat registry
在处理avro数据时会多写一个MAGIC_BYTE，一般avro是没有的，消费时用ConfluentRegistryAvroDeserializationSchema
试试。
Best，
Leonard Xu
[1]
https://is
Flink支持把Timestamp(3)这种类型声明声明为事件时间列，并且为它生成watermark。
你上面提到的"-MM-dd'T'HH:mm:ss.SSS'Z'"，并不是一种数据类型，它只是Timestamp的一种string表达形式，这个主要是在json
format里面把一个字符串解析为timestamp类型的时候需要的一种格式。
所以如果你有其他类型的字段，比如varchar、long、int等等，都可以通过内置函数或者udf将其转成timestamp(3)的类型，再在此基础上做watermark生成。
guaishushu1...@163.com 于2020年5月
我的理解是这样的。
TableEnvironment的api主要是跟table相关的概念，这里面并没有流(DataStream)和批(DataSet)的概念。
StreamTableEnvironment是继承了TableEnvironment，但是多了跟DataStream交互的接口，比如把DataStream转成Table，以及反过来转换等。
BatchTableEnvironment也是同理，添加了跟DataSet打交道的一些接口。
LakeShen 于2020年5月29日周五 上午10:16写道：
> Hi Benchao,
>
> TableEnvironment 和 Str
而且 flink不是只支持这种"-MM-dd'T'HH:mm:ss.SSS'Z'" 类型解析为watermark吗，就对这样有点疑惑
guaishushu1...@163.com
发件人： guaishushu1...@163.com
发送时间： 2020-05-29 10:20
收件人： Benchao Li
抄送： user-zh
主题： Re: Re: flink-sql watermark问题
就是我是long类型的时间戳，但是用TO_TIMESTAMP转换成'-MM-dd HH:mm:ss' 之后依然可以生成watermark。
guaishushu
更新一下这个问题的进展：
目前java 8u252的修复已经merge到了master和release-1.11分支，你可以用这两个
分支自己编译flink binary进行验证
另外，如果确实想在在1.10使用，可以设置环境变量HTTP2_DISABLE=true
Flink client端可以export HTTP2_DISABLE=true
JM/TM可以通过如下Flink参数设置，当然也可以直接在build镜像的时候设置
containerized.master.env.HTTP2_DISABLE=true
containerized.taskmanager.env.HTTP
就是我是long类型的时间戳，但是用TO_TIMESTAMP转换成'-MM-dd HH:mm:ss' 之后依然可以生成watermark。
guaishushu1...@163.com
发件人： Benchao Li
发送时间： 2020-05-28 17:00
收件人： user-zh
主题： Re: flink-sql watermark问题
Hi，
没太看明白你的问题是什么。目前的确是只支持Timestmap(3)作为事件时间列。
之所以还不支持long作为事件时间列，主要考虑的是时区问题。但是这个社区也在考虑，可以参考[1]
[1] https://issu
Hi Benchao,
TableEnvironment 和 StreamTableEnvironment 具体有什么差异吗，我看StreamTableEnvironment
继承了 TableEnvironment。
这块我不是很了解，有什么文档介绍吗，感谢。
Best,
LakeShen
Benchao Li 于2020年5月28日周四 下午5:52写道：
> Hi zhisheng&wind，
>
> 这块内容在FLIP-84[1]里面做了大量的重构和优化，而且大部分工作在1.11已经完成了，你们可以看下。
>
> [1]
> https://cwiki.apache.or
hello
正在尝试中，感谢解答🤗
best wishes
111 于2020年5月28日周四 上午10:16写道：
> Hi,
> 想要在sqlgateway里面使用，那么可以看看下面几个条件：
> 1 满足SPI的要求，能让flink自动发现实现类
> 2 配置FLINK_HOME环境变量，自定义的connector放在FLINK_HOME/lib下
> 3 如果与Hive集成，使用hivecatalog，那么先要注册表
> 这样就可以使用了。
> Best,
> Xinghalo
我使用avro代码生成工具基于一个schema生成了UserView类，又实现了kafka的Serializer接口，代码如下：
import org.apache.kafka.common.serialization.Serializer;
import java.io.IOException;
public class UserViewSerializer implements Serializer {
@Override
public byte[] serialize(String topic, UserView data) {
byte[] arr
这不是flink的问题，之前遇到过相关问题 hbase region分裂的时候 会出现此类问题 你可以看看日志是否hbase region当时正在分裂，
在 2020-05-28 16:57:35，"air23" 写道：
2020-05-28 16:54:23,867 INFO
org.apache.hadoop.hbase.client.AsyncRequestFutureImpl - id=2,
table=GC_SCHEM:mon1, attempt=7/16, failureCount=427ops, last
可以先在之前对于long类型的字段转换成Timestmap 类型 再生成watermark
在 2020-05-28 17:00:53，"Benchao Li" 写道：
>Hi，
>
>没太看明白你的问题是什么。目前的确是只支持Timestmap(3)作为事件时间列。
>之所以还不支持long作为事件时间列，主要考虑的是时区问题。但是这个社区也在考虑，可以参考[1]
>
>[1] https://issues.apache.org/jira/browse/FLINK-16938
>
>guaishushu1...@163.com 于2020年5月28日周
Hi zhisheng&wind，
这块内容在FLIP-84[1]里面做了大量的重构和优化，而且大部分工作在1.11已经完成了，你们可以看下。
[1]
https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=134745878
wind.fly@outlook.com 于2020年5月28日周四 下午5:45写道：
> Hi,
>
> StreamTableEnvironment类似于TableEnvironment的Dag优化有版本规划吗？未来TableEnvironment的dag优化和Strea
感谢两位大佬的回复，期待native kubernetes更多的特性出现
| |
a511955993
|
|
邮箱：a511955...@163.com
|
签名由 网易邮箱大师 定制
在2020年05月28日 17:39，Yang Wang 写道：
您好，
目前native方式还没有办法来挂载volume，包括hostpath、persistent volume等
这个目前已经有了JIRA ticket[1]，但是还没有开始做
如果你感兴趣，可以参与进来一起
[1]. https://issues.apache.org/jira/browse/FLINK-15649
Hi,
StreamTableEnvironment类似于TableEnvironment的Dag优化有版本规划吗？未来TableEnvironment的dag优化和StreamTableEnvironment较丰富api会不会统一？
Best,
Junbao Zhang
发件人: Benchao Li
发送时间: 2020年5月28日 17:35
收件人: user-zh
主题: Re: 疑问：flink sql 当一个env下多个insert语句共享一个view时，是否会影响view中kafka源表的offset提交
Hi，Benchao
http://zhisheng-blog.oss-cn-hangzhou.aliyuncs.com/2020-05-28-093940.jpg
这张图里面说的 TableEnvironment 不支持 UDAF/UDTF，那么如果想要用的话暂时有什么解决方法吗？社区大概什么时候会支持？
Thanks！
Benchao Li 于2020年5月28日周四 下午5:35写道：
> Hi,
>
> 时间差了13个小时这个比较奇怪，我理解不应该出现这个问题的。你是说不用TableEnvironment就不会出现这个问题么？
>
> 第二个问题，TableEnviron
您好，
目前native方式还没有办法来挂载volume，包括hostpath、persistent volume等
这个目前已经有了JIRA ticket[1]，但是还没有开始做
如果你感兴趣，可以参与进来一起
[1]. https://issues.apache.org/jira/browse/FLINK-15649
Best,
Yang
Yangze Guo 于2020年5月28日周四 下午4:11写道：
> 您好，我来回来一下第一个大问题
>
> 1.
> resources.requests.cpu和resources.limits.cpu都会被设置为kubernet
Hi,
时间差了13个小时这个比较奇怪，我理解不应该出现这个问题的。你是说不用TableEnvironment就不会出现这个问题么？
第二个问题，TableEnvironment目前应该是没法设置checkpoint的，这个应该是只能在StreamExecutionEnvironment来配置吧。
wind.fly@outlook.com 于2020年5月28日周四 下午5:27写道：
> Hi, Benchao：
>
> DAG图是指向了同一个问题，但是用了TableEnvironment后，发现所有的时间都差了13个小时，比如调用LOCALTIMESTAMP函数。另外问一
Hi, Benchao：
DAG图是指向了同一个问题，但是用了TableEnvironment后，发现所有的时间都差了13个小时，比如调用LOCALTIMESTAMP函数。另外问一个问题，使用TableEnvironment怎么设置checkpoint?
Best,
Junbao Zhang
发件人: Benchao Li
发送时间: 2020年5月28日 17:05
收件人: user-zh
主题: Re: 疑问：flink sql 当一个env下多个insert语句共享一个view时，是否会影响view中kaf
Hi
> org.apache.hadoop.hbase.RegionTooBusyException
这异常信息看起来像hbase集群在大量写入时碰到了问题，不像是flink这边的问题，建议排查下hbase侧，应该有一些参数可以优化。
Best,
Leonard Xu
嗯，是的。这个是blink planner特有的优化，可以支持多个sink的DAG图的优化。你也可以看下Flink
UI上的任务DAG图，比较明显的可以看到只有一个Source，后面接了两个Sink。
wind.fly@outlook.com 于2020年5月28日周四 下午5:02写道：
> Hi, Benchao：
> 谢谢回复，经过测试确实是这样，也就是同一个TableEnvironment下，如果有多条insert语句访问同一个表，该表的数据只消费一次？
>
>
>
>
> Best,
> Junbao Zhang
> _
Hi, Benchao：
谢谢回复，经过测试确实是这样，也就是同一个TableEnvironment下，如果有多条insert语句访问同一个表，该表的数据只消费一次？
Best,
Junbao Zhang
发件人: Benchao Li
发送时间: 2020年5月28日 15:59
收件人: user-zh
主题: Re: 疑问：flink sql 当一个env下多个insert语句共享一个view时，是否会影响view中kafka源表的offset提交
如果你的tEnv用的是TableEnvironment，而
Hi，
没太看明白你的问题是什么。目前的确是只支持Timestmap(3)作为事件时间列。
之所以还不支持long作为事件时间列，主要考虑的是时区问题。但是这个社区也在考虑，可以参考[1]
[1] https://issues.apache.org/jira/browse/FLINK-16938
guaishushu1...@163.com 于2020年5月28日周四 下午4:22写道：
> flink-1.10 sql只支持 timestamp（3） 类型字段生成watermark
> 但是long这样转换后也可以生成watermark很奇怪？
> CREATE TABLE us
2020-05-28 16:54:23,867 INFO
org.apache.hadoop.hbase.client.AsyncRequestFutureImpl - id=2,
table=GC_SCHEM:mon1, attempt=7/16, failureCount=427ops, last
exception=org.apache.hadoop.hbase.RegionTooBusyException:
org.apache.hadoop.hbase.RegionTooBusyException: Above memstore limit,
re
w_ts as TO_TIMESTAMP(FROM_UNIXTIME(rowtime/1000),'-MM-dd
HH:mm:ss'),这个语句产生的就是一个timestamp的数据Flink内置函数：https://ci.apache.org/projects/flink/flink-docs-release-1.10/zh/dev/table/functions/systemFunctions.htmlFROM_UNIXTIME(numeric[,
string])：
Returns a representation of the numeric argument as a
说明一下 读取的数据还没有到今天的数据 也就是提示文件不存在的目录xxx/ds=2020-05-28
| |
王志华
|
|
a15733178...@163.com
|
签名由网易邮箱大师定制
在2020年05月28日 16:36，阿华田 写道：
使用recursive.file.enumeration开启递归读取hdfs的目录文件，但是每次数据没有读完就会报如下错误：
java.io.IOException: Error opening the InputSplit
hdfs://xxx/ds=2020-05-28/hour=15/2020-05-28_15.log.flu
使用recursive.file.enumeration开启递归读取hdfs的目录文件，但是每次数据没有读完就会报如下错误：
java.io.IOException: Error opening the InputSplit
hdfs://xxx/ds=2020-05-28/hour=15/2020-05-28_15.log.flume2_idcfeature_kafkamq.tmp
[0,134217728]: File does not exist: /xxx/ds=2020-05-28/hour=15/
| |
王志华
|
|
a15733178...@163.com
|
签名
flink-1.10 sql只支持 timestamp（3） 类型字段生成watermark
但是long这样转换后也可以生成watermark很奇怪？
CREATE TABLE user_log (
response_size int,
rowtime BIGINT,
w_ts as TO_TIMESTAMP(FROM_UNIXTIME(rowtime/1000),'-MM-dd HH:mm:ss'),
WATERMARK FOR w_ts AS w_ts - INTERVAL '5' SECOND --5秒的延迟
)
guaishushu1...@163.com
您好，我来回来一下第一个大问题
1. resources.requests.cpu和resources.limits.cpu都会被设置为kubernetes.jobmanager.cpu
2. external-resource..kubernetes.config-key
是为1.11的新特性扩展资源框架[1]而加入的。请不要使用它来配置cpu和memory。
[1] https://issues.apache.org/jira/browse/FLINK-17044
Best,
Yangze Guo
On Thu, May 28, 2020 at 3:48 PM wrote:
如果你的tEnv用的是TableEnvironment，而不是StreamTableEnvironment的话，两个insert会公用前面的source，也就是会只读取a表一次，然后分别给下游c和d用。
wind.fly@outlook.com 于2020年5月28日周四 下午3:14写道：
> Hi,all:
> 当前使用版本flink 1.10.0，使用blink planner，假如有如下代码：
>
> tEnv.createTemporaryView("b", tEnv.sqlQuery("select * from a"));
> tEnv.sqlUpdate("in
hi all
我在使用native kubernetes的时候，对几个配置项有疑问，想得到解答。
1. kubernetes.jobmanager.cpu配置项针对一个TM配置多少个cpu资源，是否在resources.requests.cpu 或者
resources.limits.cpu也做了配置？
在https://ci.apache.org/projects/flink/flink-docs-master/ops/config.html#kubernetes上看到对应的一个新的配置external-resource..kubernetes.config-key。
对exte
回复的好详细！而且引出了相关的测试用例
Thanks very much！
在 2020-05-28 14:23:33，"Leonard Xu" 写道：
>
>>|INSERT INTO dwdCatalog.dwd.t1_copy partition (`p_year` = p_year,
>> `p_month` = p_month)
>>|select id,name from dwdCatalog.dwd.t1 where `p_year` = 2020 and
>> `p_month` = 4
>
>动态分
Hi,all:
当前使用版本flink 1.10.0，使用blink planner，假如有如下代码：
tEnv.createTemporaryView("b", tEnv.sqlQuery("select * from a"));
tEnv.sqlUpdate("insert into c select * from b where b.con1 = '1'");
tEnv.sqlUpdate("insert into d select * from b where b.con1 = '2'");
其中a是kafka表，connector属性为：
'connector.properti
43 matches
Mail list logo
