回复： Flink Application + HA与HistoryServer的使用问题

thanks，Yang. Btw, 恭喜晋升apache flink的PMC！ Best， tanjialiang. 在2022年5月12日 10:32，Yang Wang 写道： 可以临时通过-D "$internal.pipeline.job-id="来自定义job id，但是个内部参数 你可以看下[1]，了解更多讨论的信息 [1]. https://issues.apache.org/jira/browse/FLINK-19358 Best, Yang 谭家良 于2022年5月11日周三 22:17写道：

Re:Re: How can I set job parameter in flink sql

Ok, got it. Thanks so much！ Regards, Hunk -- 发自我的网易邮箱手机智能版 在 2022-05-11 16:46:14，yuxia 写道： Hi, AFAK, you can't get the parameter setted via Flink SQL client in udf. If you still want to get the parameters in your udf, you can use the following code to set the parameter: env =

Re:Re: How can I set job parameter in flink sql

Ok, got it. Thanks so much！ Regards, Hunk -- 发自我的网易邮箱手机智能版 在 2022-05-11 16:46:14，yuxia 写道： Hi, AFAK, you can't get the parameter setted via Flink SQL client in udf. If you still want to get the parameters in your udf, you can use the following code to set the parameter: env =

Re: How to get flink to use POJO serializer when enum is present in POJO class

Hi Arvid, Thanks for replying. But I have all the getters and setters in the example. As you can see, the val2 field is commented and hence its getter and setter are commented out. When restoring from a savepoint, I uncomment these and get errors. If I remove reference to the enum RuleType from

Configuring heap size for Flink client

Hi guys, I'm developing a submission platform for Flink jobs deployed in containers using Flink client for submission. I found that the Flink client by default uses the default JVM heap setting (with an initial heap size of 1/64 host memory, and a max heap size of 1/4 host memory). When the

http stream as input data source

Hi, is there a http data stream as data source ? I only see socketTextStream and other predefined stream sources. It seems that I have to use fromCollection, fromElements ... and prepare the collection for myself. Thanks Regards

AWS EMR Yarn api shutdown flink task container does`t work

hi all : now I use AWS EMR on EC2 , and I running flink application on EMR。 sometimes I want resize task number to save on budget。 when I request resize task number ,the node can`t dimss because some flink task runing on it. so I using some yarn api to kill this contaier which runing task

Re: [Discuss] Creating an Apache Flink slack workspace

Hi all, I think forum might be a good choice for search and maintain. However, unlike slack workspace, it seems no existing popular product could be leveraged easily. Thus, I am +1 to create an Apache Flink slack channel. If the ASF slack cannot be joined easily for most of users, I prefer to

转发：基于flink sql作业失败与取消状况下的作业参数设置咨询(flink-1.14.2)

| | Jerry Guo | | wangyixuhongm...@163.com | 转发的原邮件 | 发件人 | 徐战辉 | | 发送日期 | 2022年5月12日 10:38 | | 收件人 | user-zh@flink.apache.org | | 主题 | 基于flink sql作业失败与取消状况下的作业参数设置咨询(flink-1.14.2) |

Re: [Discuss] Creating an Apache Flink slack workspace

Hi all, Regarding using ASF slack. I share the problems I saw in the Apache Druid community. [1] > As you may have heard, it’s become increasingly difficult for new users without an @apache.org email address to join the ASF #druid Slack channel. ASF Infra disabled the option to publicly provide

Re: Migrating Flink apps across cloud with state

Thank you for the suggestions, guys! @Andrew Otto This is the way we will most likely go. However, this will require us to meddle with the Flink consumer codebase. And looks like there is no other way around it. We will add some custom code to perform offset resetting for specific savepoints.

基于flink sql作业失败与取消状况下的作业参数设置咨询(flink-1.14.2)

hello, 请教下，如何设置flink配置及作业参数，在取消作业重新部署、flink作业失败重跑情况下，保证不丢失数据。 目前有一份作业，开启checkpoint, cancel 后重新启动，发现数据会丢失1小部分。 1. flink.conf execution.checkpointing.interval: 1 execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION

Re: Flink on Native K8s jobs turn in to `SUSPENDED` status unexpectedly.

The SUSPENDED state is usually caused by lost leadership. Maybe you could find more information about leader in the JobManager and TaskManager logs. Best, Yang Xiaolong Wang 于2022年5月11日周三 19:18写道： > Hello, > > Recently our Flink jobs on Native K8s encountered failing in the > `SUSPENDED`

Re: Flink Application + HA与HistoryServer的使用问题

可以临时通过-D "$internal.pipeline.job-id="来自定义job id，但是个内部参数 你可以看下[1]，了解更多讨论的信息 [1]. https://issues.apache.org/jira/browse/FLINK-19358 Best, Yang 谭家良 于2022年5月11日周三 22:17写道： > > > 我使用的Application模式：Kubernetes > 我使用的HA模式：Kubernetes HA > > > 目前Application + HA发现所有的Job

Source without persistent state

Hello, I'm working on custom Source, something like heartbeat generator using new Source API, HeartSource is constructed with list of Kafka topics, SplitEnumerator for each topic queries number of partitions, and either creates a split per topic-partition or single split for all

Re: [Discuss] Creating an Apache Flink slack workspace

> > To make some progress, maybe we decide on chat vs forum vs none and then > go into a deeper discussion on the implementation or is there anything > about Slack that would be complete blocker for the implementation? > Sure, then I'd be +1 for chat. From my side, the initiative is more about

Re: unsubscribe

To unsubscribe, you can send email to user-unsubscr...@flink.apache.org with any object. Best regards, Yuxia 发件人: "Henry Cai" 收件人: "User" 发送时间: 星期四, 2022年 5 月 12日 上午 1:14:43 主题: unsubscribe unsubscribe

Checkpointing - Job Manager S3 Head Requests are 404s

We're using S3 to store checkpoints. They are taken every minute. I'm seeing a large number of 404 responses from S3 being generated by the job manager. The order of the entries in the debugging log would imply that it's a result of a HEAD request to a key. For example all the incidents look like

Incompatible data types while using firehose sink

Hi Folks, Getting this error when sinking data to a firehosesink, would really appreciate some help ! DataStream inputStream = env.addSource(new FlinkKafkaConsumer<>("xxx", new SimpleStringSchema(), properties)); Properties sinkProperties = new Properties();

Incompatible data types while using firehose sink

Hi Folks, Getting this error when sinking data to a firehosesink, would really appreciate some help ! DataStream inputStream = env.addSource(new FlinkKafkaConsumer<>("xxx", new SimpleStringSchema(), properties)); Properties sinkProperties = new Properties();

How to define TypeInformation for Flink recursive resolved POJO

Hi Community, I have a POJO that has nested recursively resolved structure. How should I define the @TypeInfo annotation correctly to avoid stack overflow exception when starting the application. Basically, Class Metadata Map fields Class FieldDefinition Metadata parentMetadata The Metadata

unsubscribe

unsubscribe

Re: Practical guidance with Scala and Flink >= 1.15

As a yet another attempt to make Flink work with scala 2.13/3.x, we went further and and cross-built a forked version of Flink's Scala API: https://github.com/findify/flink-scala-api Check the github repo for details, but if you can afford

[QUESTION] In Flink k8s Application mode with HA can not using History Server for history backend

In Flink k8s application mode with high-availability, it's job id always 00, but in history server, it make job's id for the key. How can I using the application mode with HA and store the history job status with history server? Best, tanjialiang.

Re: Resizing kube container sizes dynamically for custom jobs

Hi Morgan, Jobs running in a session cluster share the taskmanagers, so you are not able to configure them on a per job basis. I welcome you to check out the Flink Kubernetes Operator's session job example [1] that highlights this behavior: You specify container resources when you submit the

Flink Application + HA与HistoryServer的使用问题

我使用的Application模式：Kubernetes 我使用的HA模式：Kubernetes HA 目前Application + HA发现所有的Job ID都变成了“”，但是在HistoryServer中，他会使用Job ID作为归档文件的文件名。那么如果这样，HistoryServer应该怎么与Application + HA相结合呢？

Resizing kube container sizes dynamically for custom jobs

Greetings all, I have a question concerning resource allocation for Apache flink. I have a flink native session cluster running and im interested in rolling out multiple jobs. However, I would like to size the container resources (CPU and Memory) differently for each job, is this possible?

Flink on Native K8s jobs turn in to `SUSPENDED` status unexpectedly.

Hello, Recently our Flink jobs on Native K8s encountered failing in the `SUSPENDED` status and got restarted for no reason. Flink version: 1.13.2 Logs: ``` 2022-05-11 05:01:41 2022-05-10 21:01:41,771 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator [] - Triggering checkpoint

Re: [Discuss] Creating an Apache Flink slack workspace

I don't think we can maintain two additional channels. Some people have already concerns about covering one additional channel. I think, a forum provides a better user experience than a mailing list. Information is structured better, you can edit messages, sign up and search is easier. To make

Re: How can I set job parameter in flink sql

Hi, AFAK, you can't get the parameter setted via Flink SQL client in udf. If you still want to get the parameters in your udf, you can use the following code to set the parameter: env = StreamExecutionEnvironment.getExecutionEnvironment parameter = new HashMap(); parameter .put("

Re: How can I set job parameter in flink sql

Hi, AFAK, you can't get the parameter setted via Flink SQL client in udf. If you still want to get the parameters in your udf, you can use the following code to set the parameter: env = StreamExecutionEnvironment.getExecutionEnvironment parameter = new HashMap(); parameter .put("

Re: OOM errors cause by the new KafkaSink API

Hi Hua Wei, Thanks for the investigation. Could you provide a heap dump before the crash? The OOM stacktrace that you are showing is rather random (at RPC message exchange). We need to see where the heap is growing. Alternatively, you can take heap dumps at different points in time and compare

Re: Practical guidance with Scala and Flink >= 1.15

Sorry, I forgot the link of the repo: https://github.com/salvalcantara/flink-scala Regards, Salva On Wed, May 11, 2022 at 9:32 AM Salva Alcántara wrote: > Thanks Martijn, my conclusion so far is that Java is a safe bet. > > In the meantime, a friend and I have spent some time trying to make

Re: Practical guidance with Scala and Flink >= 1.15

Hi, guys. I posted a jdk11 & jdk17 issue [FLINK-27549] recently which involved upgrading scala [2] of current discussion. It shows that the current flink project is not a completed or pure jdk11 version.(same problem with higher version).

Re: Practical guidance with Scala and Flink >= 1.15

Thanks Martijn, my conclusion so far is that Java is a safe bet. In the meantime, a friend and I have spent some time trying to make `flink-scala` work with Flink 1.15 and Scala 2.13. We partly followed the discussions in [1] (FLINK-13414) to fix all the compilations errors. Note that this is

Re: flink sql无法读取Hive映射的HBase表

不好意思，我尝试复现你的问题，但是我没有 hbase 环境，不过看起来是只有当 STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler' 有问题？ 我之后空了再debug 看看。 不过我看了一下 flink 这块的代码，从 flink 这块的代码来看，应该是 get 这个 hive 表之后，它的 StorageDescriptor 的 inputformat 为 null，然后 Class.forName(inputformat) 就报错 NPE了。 应该是这块代码有点问题。 如果你方便的话，可以辛苦帮忙建一个

Query on using Apache Flink with Confluent Kafka

Hi Team, I have a requirement to read kafka events through Apache Flink and do processing with the same. Now this kafka topic which produces the event to the Apache Flink is a confluent kafka and it is hosted as a kubernetes pod in the Docker container. The actual problem is I am unable to

Re: [ANNOUNCE] Apache Flink Table Store 0.1.0 released

Really excited to see the very first release of the flink-table-store! Kudos to everyone who helped with this effort! Cheers, Jiangjie (Becket) Qin On Wed, May 11, 2022 at 1:55 PM Jingsong Lee wrote: > The Apache Flink community is very happy to announce the release of Apache > Flink Table

Re: [ANNOUNCE] Apache Flink Table Store 0.1.0 released

Really excited to see the very first release of the flink-table-store! Kudos to everyone who helped with this effort! Cheers, Jiangjie (Becket) Qin On Wed, May 11, 2022 at 1:55 PM Jingsong Lee wrote: > The Apache Flink community is very happy to announce the release of Apache > Flink Table

Re: Practical guidance with Scala and Flink >= 1.15

Hi Matthias, Given the current state of Scala support in the Flink community (there is a major lack in Scala maintainers), it is my personal opinion that we should consider deprecating the current Scala APIs and replace those with new Scala APIs, which are 'just' wrappers for the Java API. This

Re: Getting JSON kafka source parsing error with DDL

Never mind. Figured out. Wrong connector arguments. On Tue, May 10, 2022 at 11:19 PM Shubham Bansal < shubham.bansal2...@gmail.com> wrote: > Hi Everyone, > > I am trying to fix the flink-playground for version 1.14.4 and was working > on fixing pyflink-walkthrough and I getting following error >

How can I set job parameter in flink sql

Hi dear engineer, I want to override the function open() in my UDF, like: | public class BlackListConvertFunction extends ScalarFunction { @Override public void open(FunctionContext context) throws Exception { String path = context.getJobParameter("black_list_path",

How can I set job parameter in flink sql

Hi dear engineer, I want to override the function open() in my UDF, like: | public class BlackListConvertFunction extends ScalarFunction { @Override public void open(FunctionContext context) throws Exception { String path = context.getJobParameter("black_list_path",

How can I set job parameter in flink sql

Hi dear engineer, I want to override the function open() in my UDF, like: In open() function, I want to fetch the configred value "black_list_path", then simply print that value out. And I config this value in ./sql-client.sh console: SET black_list_path = /root/list.properties Then

How can I set job parameter in flink sql

Hi dear engineer, I want to override the function open() in my UDF, like: In open() function, I want to fetch the configred value "black_list_path", then simply print that value out. And I config this value in ./sql-client.sh console: SET black_list_path = /root/list.properties Then

Getting JSON kafka source parsing error with DDL

Hi Everyone, I am trying to fix the flink-playground for version 1.14.4 and was working on fixing pyflink-walkthrough and I getting following error Caused by: org.apache.flink.table.api.NoMatchingTableFactoryException: Could not find a suitable table factory for

Re: Converting from table to stream, following Avro schema

Hi Dhavan, The type of the `ds` appearing in `t_env.from_data_stream(ds) should be known. Otherwise, it's impossible to infer the schema of the converted table, as a result, `raw` type will be used which makes the schema of the resulting table not expected. You could either declare the type of