Setting flink-conf params in IDE

[Alexandru Gutan]

Hi everyone!

Is there a way to set flink-conf.yaml params but when running from the IDE?

What I'm trying to do is to setup JMX metrics:

metrics.reporter.jmx.class:
org.apache.flink.metrics.jmx.JMXReportermetrics.reporter.jmx.port:
8789

Thanks!

Re: Parallelism questions

[Alexandru Gutan]

Thats great news!

Are there any plans to expose it in the upcoming Flink release?

On Tue, 15 Jan 2019 at 12:59, Till Rohrmann  wrote:

> Hi Alexandru,
>
> at the moment `/jobs/:jobid/rescaling` will always change the parallelism
> for all operators. The maximum is the maximum parallelism which you have
> defined for an operator.
>
> I agree that it should also be possible to rescale an individual operator.
> There internal functionality is already implemented (see
> JobMaster#rescaleOperators) but has not been exposed.
>
> Cheers,
> Till
>
> On Tue, Jan 15, 2019 at 1:03 PM Alexandru Gutan 
> wrote:
>
>> Thanks Till!
>>
>> To execute the above (using Kubernetes), one would enter the running
>> JobManager service and execute it?
>> The following REST API call does the same */jobs/:jobid/rescaling*?
>>
>> I assume it changes the base parallelism, but what it will do if I had
>> already set the parallelism of my operators?
>> e.g.
>> .source(..)
>> .setParallelism(3)
>> .setUID(..)
>> .map(..)
>> .setParallelism(8)
>> .setUID(..)
>> .sink(..)
>> .setParallelism(3)
>> .setUID(..)
>>
>> I think it would be a good idea to have */jobs/:jobid/rescaling,* 
>> additionally
>> requiring the *operatorUID* as a queryParameter*, *so that the
>> parallelism of specific operators could be changed.
>>
>> Best,
>> Alex.
>>
>> On Tue, 15 Jan 2019 at 10:27, Till Rohrmann  wrote:
>>
>>> Hi Alexandru,
>>>
>>> you can use the `modify` command `bin/flink modify 
>>> --parallelism ` to modify the parallelism of a job. At the
>>> moment, it is implemented as first taking a savepoint, stopping the job and
>>> then redeploying the job with the changed parallelism and resuming from the
>>> savepoint.
>>>
>>> Cheers,
>>> Till
>>>
>>> On Mon, Jan 14, 2019 at 4:21 PM Dawid Wysakowicz 
>>> wrote:
>>>
>>>> Hi Alexandru
>>>>
>>>> As for 2, generally speaking the number of required slots depends on
>>>> number of slot sharing groups. By default all operators belong to the
>>>> default slot sharing group, that means a job requires as many slots as
>>>> maximal parallelism in the job. More on the distributed runtime you can
>>>> read here[1]
>>>>
>>>> As for 1 I cc'ed Gary and Till who might better answer your question.
>>>>
>>>> [1]
>>>> https://ci.apache.org/projects/flink/flink-docs-release-1.7/concepts/runtime.html#task-slots-and-resources
>>>>
>>>> Best,
>>>>
>>>> Dawid
>>>> On 14/01/2019 15:26, Alexandru Gutan wrote:
>>>>
>>>> Hi everyone!
>>>>
>>>> 1. Is there a way to increase the parallelism (e.g. through REST) of
>>>> some operators in a job without re-deploying the job? I found this
>>>> <https://stackoverflow.com/questions/50719147/apache-flink-guideliness-for-setting-parallelism>
>>>> answer which mentions scaling at runtime on Yarn/Mesos. Is it possible?
>>>> How? Support for Kubernetes?
>>>> 2. What happens when the number of parallel operator instances exceeds
>>>> the number of task slots? For example: a job with a source (parallelism 3),
>>>> a map (parallelism 8), a sink (parallelism 3), total of *14* operator
>>>> instances and a setup with *8* task slots. Will the operators get
>>>> chained? What if I disable operator chaining?
>>>>
>>>> Thank you!
>>>>
>>>>

Re: Re: How can I make HTTP requests from an Apache Flink program?

[Alexandru Gutan]

Hi Jacopo,

Check this:
https://ci.apache.org/projects/flink/flink-docs-release-1.7/dev/stream/operators/asyncio.html

Best,
Alex

On Tue, 15 Jan 2019 at 13:57,  wrote:

> Hi,
>
>
>
> I have a flink program which needs to process many messages and part of
> this processing is to process the data using an external web service using
> http calls.
>
> Example:
>
> *val *myStream: DataStream[String]
> myStream
>   .map(*new *MyProcessingFunction)
>   .map(*new *MyWebServiceHttpClient)
>   .print
>
> Any suggestion?
>
>
>
> Jac
>
>
>
>
>
> *From:* miki haiat [mailto:miko5...@gmail.com]
> *Sent:* Dienstag, 15. Januar 2019 14:03
> *To:* Gobbi, Jacopo-XT
> *Cc:* user
> *Subject:* [External] Re: How can I make HTTP requests from an Apache
> Flink program?
>
>
>
> Can you share more which use case are you trying to implement ?
>
>
>
>
>
>
>
> On Tue, Jan 15, 2019 at 2:02 PM  wrote:
>
> Hi all,
>
>
>
> I was wondering if anybody has any recommendation over making HTTP
> requests from Flink to another service.
>
> On the long term we are looking for a solution that is both performing and
> integrates well with our flink program.
>
> Does it matter the library we use? Do we need a special connector to make
> HTTP calls?
>
> One library we thought that could fit our necessities Akka akka HTTP
> client API due to the possibility to make async HTTP calls.
>
>
>
> We are using Scala 2.12 and Flink 1.7.
>
>
>
> Kind regards,
>
>
>
> Jacopo Gobbi
>
>

Re: Parallelism questions

[Alexandru Gutan]

Thanks Till!

To execute the above (using Kubernetes), one would enter the running
JobManager service and execute it?
The following REST API call does the same */jobs/:jobid/rescaling*?

I assume it changes the base parallelism, but what it will do if I had
already set the parallelism of my operators?
e.g.
.source(..)
.setParallelism(3)
.setUID(..)
.map(..)
.setParallelism(8)
.setUID(..)
.sink(..)
.setParallelism(3)
.setUID(..)

I think it would be a good idea to have */jobs/:jobid/rescaling,* additionally
requiring the *operatorUID* as a queryParameter*, *so that the parallelism
of specific operators could be changed.

Best,
Alex.

On Tue, 15 Jan 2019 at 10:27, Till Rohrmann  wrote:

> Hi Alexandru,
>
> you can use the `modify` command `bin/flink modify  --parallelism
> ` to modify the parallelism of a job. At the moment, it is
> implemented as first taking a savepoint, stopping the job and then
> redeploying the job with the changed parallelism and resuming from the
> savepoint.
>
> Cheers,
> Till
>
> On Mon, Jan 14, 2019 at 4:21 PM Dawid Wysakowicz 
> wrote:
>
>> Hi Alexandru
>>
>> As for 2, generally speaking the number of required slots depends on
>> number of slot sharing groups. By default all operators belong to the
>> default slot sharing group, that means a job requires as many slots as
>> maximal parallelism in the job. More on the distributed runtime you can
>> read here[1]
>>
>> As for 1 I cc'ed Gary and Till who might better answer your question.
>>
>> [1]
>> https://ci.apache.org/projects/flink/flink-docs-release-1.7/concepts/runtime.html#task-slots-and-resources
>>
>> Best,
>>
>> Dawid
>> On 14/01/2019 15:26, Alexandru Gutan wrote:
>>
>> Hi everyone!
>>
>> 1. Is there a way to increase the parallelism (e.g. through REST) of some
>> operators in a job without re-deploying the job? I found this
>> <https://stackoverflow.com/questions/50719147/apache-flink-guideliness-for-setting-parallelism>
>> answer which mentions scaling at runtime on Yarn/Mesos. Is it possible?
>> How? Support for Kubernetes?
>> 2. What happens when the number of parallel operator instances exceeds
>> the number of task slots? For example: a job with a source (parallelism 3),
>> a map (parallelism 8), a sink (parallelism 3), total of *14* operator
>> instances and a setup with *8* task slots. Will the operators get
>> chained? What if I disable operator chaining?
>>
>> Thank you!
>>
>>

Parallelism questions

[Alexandru Gutan]

Hi everyone!

1. Is there a way to increase the parallelism (e.g. through REST) of some
operators in a job without re-deploying the job? I found this

answer which mentions scaling at runtime on Yarn/Mesos. Is it possible?
How? Support for Kubernetes?
2. What happens when the number of parallel operator instances exceeds the
number of task slots? For example: a job with a source (parallelism 3), a
map (parallelism 8), a sink (parallelism 3), total of *14* operator
instances and a setup with *8* task slots. Will the operators get chained?
What if I disable operator chaining?

Thank you!

Re: Flink on Kubernetes (Minikube)

[Alexandru Gutan]

Got it working on the Google Cloud Platform Kubernetes service...
More support for Minikube is needed.

On Wed, 19 Dec 2018 at 13:44, Alexandru Gutan 
wrote:

> I've found this in the archives:
> http://mail-archives.apache.org/mod_mbox/flink-dev/201804.mbox/%3CCALbFKXr=rp9TYpD_JA8vmuWbcjY0+Lp2mbr4Y=0fnh316hz...@mail.gmail.com%3E
>
> And as suggested I tried a different startup order but unsuccessful:
>
> kubectl create -f jobmanager-deployment.yaml
> kubectl create -f jobmanager-service.yaml
> kubectl create -f taskmanager-deployment.yaml
>
> I get the same error *java.net.UnknownHostException: flink-jobmanager: 
> Temporary failure in name resolution*
>
>
> On Wed, 19 Dec 2018 at 13:27, Dawid Wysakowicz 
> wrote:
>
>> Hi Alexandru,
>>
>> This sounds reasonable that it might be because of this minikube command
>> failed, but I am not a kubernetes expert. I cc Till who knows more on this.
>>
>> Best,
>>
>> Dawid
>> On 19/12/2018 14:16, Alexandru Gutan wrote:
>>
>> Thanks!
>> I'm using now the *flink:1.7.0-hadoop24-scala_2.12* image.
>> The Hadoop related error is gone, but I have a new error:
>>
>> Starting Task Manager
>> config file:
>> jobmanager.rpc.address: flink-jobmanager
>> jobmanager.rpc.port: 6123
>> jobmanager.heap.size: 1024m
>> taskmanager.heap.size: 1024m
>> taskmanager.numberOfTaskSlots: 2
>> parallelism.default: 1
>> rest.port: 8081
>> blob.server.port: 6124
>> query.server.port: 6125
>> Starting taskexecutor as a console application on host
>> flink-taskmanager-54b679f8bb-22b4r.
>> 2018-12-19 13:09:38,469 INFO
>> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -
>> 
>> 2018-12-19 13:09:38,470 INFO
>> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  Starting
>> TaskManager (Version: 1.7.0, Rev:49da9f9, Date:28.11.2018 @ 17:59:06 UTC)
>> 2018-12-19 13:09:38,470 INFO
>> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  OS current
>> user: flink
>> 2018-12-19 13:09:38,921 WARN
>> org.apache.hadoop.util.NativeCodeLoader   - Unable to
>> load native-hadoop library for your platform... using builtin-java classes
>> where applicable
>> 2018-12-19 13:09:39,307 INFO
>> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  Current
>> Hadoop/Kerberos user: flink
>> 2018-12-19 13:09:39,307 INFO
>> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  JVM:
>> OpenJDK 64-Bit Server VM - Oracle Corporation - 1.8/25.181-b13
>> 2018-12-19 13:09:39,307 INFO
>> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  Maximum
>> heap size: 922 MiBytes
>> 2018-12-19 13:09:39,307 INFO
>> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  JAVA_HOME:
>> /docker-java-home/jre
>> 2018-12-19 13:09:39,318 INFO
>> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  Hadoop
>> version: 2.4.1
>> 2018-12-19 13:09:39,318 INFO
>> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  JVM
>> Options:
>> 2018-12-19 13:09:39,319 INFO
>> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -
>> -XX:+UseG1GC
>> 2018-12-19 13:09:39,319 INFO
>> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   - -Xms922M
>> 2018-12-19 13:09:39,320 INFO
>> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   - -Xmx922M
>> 2018-12-19 13:09:39,320 INFO
>> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -
>> -XX:MaxDirectMemorySize=8388607T
>> 2018-12-19 13:09:39,320 INFO
>> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -
>> -Dlog4j.configuration=file:/opt/flink/conf/log4j-console.properties
>> 2018-12-19 13:09:39,320 INFO
>> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -
>> -Dlogback.configurationFile=file:/opt/flink/conf/logback-console.xml
>> 2018-12-19 13:09:39,320 INFO
>> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  Program
>> Arguments:
>> 2018-12-19 13:09:39,321 INFO
>> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -
>> --configDir
>> 2018-12-19 13:09:39,321 INFO
>> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -
>> /opt/flink/conf
>> 2018-12-19 13:09:39,321 INFO
>> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  Classpath:
>> /opt/flink/lib/flink-python_2.12-1.7.0.jar:/opt/flink/lib/flink-shaded-hadoop2-uber-1.7.0.jar:/opt/f

Re: Flink on Kubernetes (Minikube)

[Alexandru Gutan]

I've found this in the archives:
http://mail-archives.apache.org/mod_mbox/flink-dev/201804.mbox/%3CCALbFKXr=rp9TYpD_JA8vmuWbcjY0+Lp2mbr4Y=0fnh316hz...@mail.gmail.com%3E

And as suggested I tried a different startup order but unsuccessful:

kubectl create -f jobmanager-deployment.yaml
kubectl create -f jobmanager-service.yaml
kubectl create -f taskmanager-deployment.yaml

I get the same error *java.net.UnknownHostException: flink-jobmanager:
Temporary failure in name resolution*


On Wed, 19 Dec 2018 at 13:27, Dawid Wysakowicz 
wrote:

> Hi Alexandru,
>
> This sounds reasonable that it might be because of this minikube command
> failed, but I am not a kubernetes expert. I cc Till who knows more on this.
>
> Best,
>
> Dawid
> On 19/12/2018 14:16, Alexandru Gutan wrote:
>
> Thanks!
> I'm using now the *flink:1.7.0-hadoop24-scala_2.12* image.
> The Hadoop related error is gone, but I have a new error:
>
> Starting Task Manager
> config file:
> jobmanager.rpc.address: flink-jobmanager
> jobmanager.rpc.port: 6123
> jobmanager.heap.size: 1024m
> taskmanager.heap.size: 1024m
> taskmanager.numberOfTaskSlots: 2
> parallelism.default: 1
> rest.port: 8081
> blob.server.port: 6124
> query.server.port: 6125
> Starting taskexecutor as a console application on host
> flink-taskmanager-54b679f8bb-22b4r.
> 2018-12-19 13:09:38,469 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -
> 
> 2018-12-19 13:09:38,470 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  Starting
> TaskManager (Version: 1.7.0, Rev:49da9f9, Date:28.11.2018 @ 17:59:06 UTC)
> 2018-12-19 13:09:38,470 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  OS current
> user: flink
> 2018-12-19 13:09:38,921 WARN
> org.apache.hadoop.util.NativeCodeLoader   - Unable to
> load native-hadoop library for your platform... using builtin-java classes
> where applicable
> 2018-12-19 13:09:39,307 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  Current
> Hadoop/Kerberos user: flink
> 2018-12-19 13:09:39,307 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  JVM:
> OpenJDK 64-Bit Server VM - Oracle Corporation - 1.8/25.181-b13
> 2018-12-19 13:09:39,307 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  Maximum
> heap size: 922 MiBytes
> 2018-12-19 13:09:39,307 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  JAVA_HOME:
> /docker-java-home/jre
> 2018-12-19 13:09:39,318 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  Hadoop
> version: 2.4.1
> 2018-12-19 13:09:39,318 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  JVM
> Options:
> 2018-12-19 13:09:39,319 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -
> -XX:+UseG1GC
> 2018-12-19 13:09:39,319 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   - -Xms922M
> 2018-12-19 13:09:39,320 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   - -Xmx922M
> 2018-12-19 13:09:39,320 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -
> -XX:MaxDirectMemorySize=8388607T
> 2018-12-19 13:09:39,320 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -
> -Dlog4j.configuration=file:/opt/flink/conf/log4j-console.properties
> 2018-12-19 13:09:39,320 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -
> -Dlogback.configurationFile=file:/opt/flink/conf/logback-console.xml
> 2018-12-19 13:09:39,320 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  Program
> Arguments:
> 2018-12-19 13:09:39,321 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -
> --configDir
> 2018-12-19 13:09:39,321 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -
> /opt/flink/conf
> 2018-12-19 13:09:39,321 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  Classpath:
> /opt/flink/lib/flink-python_2.12-1.7.0.jar:/opt/flink/lib/flink-shaded-hadoop2-uber-1.7.0.jar:/opt/flink/lib/log4j-1.2.17.jar:/opt/flink/lib/slf4j-log4j12-1.7.15.jar:/opt/flink/lib/flink-dist_2.12-1.7.0.jar:::
> 2018-12-19 13:09:39,321 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -
> 
> 2018-12-19 13:09:39,323 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   - Registered
> UNIX signal handlers for [TERM, HUP, INT]
> 2018-12-19 13:09:39,329 INFO
> org.apache.flink.runtime.taskexecutor.

Re: Flink on Kubernetes (Minikube)

[Alexandru Gutan]

: blob.server.port, 6124
2018-12-19 13:09:39,374 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: query.server.port, 6125
2018-12-19 13:09:39,511 INFO
org.apache.flink.runtime.security.modules.HadoopModule- Hadoop user
set to flink (auth:SIMPLE)
2018-12-19 13:10:00,708 ERROR
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   - TaskManager
initialization failed.



















*java.net.UnknownHostException: flink-jobmanager: Temporary failure in name
resolutionat java.net.Inet4AddressImpl.lookupAllHostAddr(Native
Method)at
java.net.InetAddress$2.lookupAllHostAddr(InetAddress.java:929)at
java.net.InetAddress.getAddressesFromNameService(InetAddress.java:1324)
at java.net.InetAddress.getAllByName0(InetAddress.java:1277)at
java.net.InetAddress.getAllByName(InetAddress.java:1193)at
java.net.InetAddress.getAllByName(InetAddress.java:1127)at
java.net.InetAddress.getByName(InetAddress.java:1077)at
org.apache.flink.runtime.rpc.akka.AkkaRpcServiceUtils.getRpcUrl(AkkaRpcServiceUtils.java:167)
at
org.apache.flink.runtime.rpc.akka.AkkaRpcServiceUtils.getRpcUrl(AkkaRpcServiceUtils.java:133)
at
org.apache.flink.runtime.highavailability.HighAvailabilityServicesUtils.createHighAvailabilityServices(HighAvailabilityServicesUtils.java:89)
at
org.apache.flink.runtime.taskexecutor.TaskManagerRunner.(TaskManagerRunner.java:127)
at
org.apache.flink.runtime.taskexecutor.TaskManagerRunner.runTaskManager(TaskManagerRunner.java:330)
at
org.apache.flink.runtime.taskexecutor.TaskManagerRunner$1.call(TaskManagerRunner.java:301)
at
org.apache.flink.runtime.taskexecutor.TaskManagerRunner$1.call(TaskManagerRunner.java:298)
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:1556)
at
org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)
at
org.apache.flink.runtime.taskexecutor.TaskManagerRunner.main(TaskManagerRunner.java:298)*

Is the minikube ssh... command supposed to mitigate this? (*minikube ssh
'sudo ip link set docker0 promisc on')*
Maybe the minikube ssh.. command didn't get executed properly? Is there a
way to check if it had executed correctly?
Or is it another type of issue?


*Thank you!*

On Wed, 19 Dec 2018 at 12:12, Dawid Wysakowicz 
wrote:

> Hi,
>
> You used a hadoopless docker image, therefore it cannot find hadoop
> dependencies. It is ok if you don't need to use any, the bolded messages
> are just INFO, those are not errors.
>
> Best,
>
> Dawid
> On 19/12/2018 12:58, Alexandru Gutan wrote:
>
> Dear all,
>
> I followed the instructions found here:
> https://ci.apache.org/projects/flink/flink-docs-release-1.7/ops/deployment/kubernetes.html
> Minikube version 0.31-01
> Kubernetes version 1.10
> Flink Docker image: flink:latest (1.7.0-scala_2.12)
>
> I ran the following commands:
>
> minikube start
> minikube ssh 'sudo ip link set docker0 promisc on'
> kubectl create -f jobmanager-deployment.yaml
> kubectl create -f taskmanager-deployment.yaml
> kubectl create -f jobmanager-service.yaml
>
> The 2 taskmanagers fail.
> Output:
>
> Starting Task Manager
> config file:
> jobmanager.rpc.address: flink-jobmanager
> jobmanager.rpc.port: 6123
> jobmanager.heap.size: 1024m
> taskmanager.heap.size: 1024m
> taskmanager.numberOfTaskSlots: 2
> parallelism.default: 1
> rest.port: 8081
> blob.server.port: 6124
> query.server.port: 6125
> Starting taskexecutor as a console application on host
> flink-taskmanager-7679c9d55d-n2trk.
> 2018-12-19 11:42:45,216 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -
> 
> 2018-12-19 11:42:45,218 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  Starting
> TaskManager (Version: 1.7.0, Rev:49da9f9, Date:28.11.2018 @ 17:59:06 UTC)
> 2018-12-19 11:42:45,218 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  OS current
> user: flink
> 2018-12-19 11:42:45,219 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  *Current
> Hadoop/Kerberos user: *
> 2018-12-19 11:42:45,219 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  JVM:
> OpenJDK 64-Bit Server VM - Oracle Corporation - 1.8/25.181-b13
> 2018-12-19 11:42:45,219 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  Maximum
> heap size: 922 MiBytes
> 2018-12-19 11:42:45,220 INFO
> org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  JAVA_HOME:
> /docker-java-home/jre
> 2018-12-19 11:42:45,220 INFO
> org.apache.flink.runtime.taskexecutor.TaskM

Flink on Kubernetes (Minikube)

[Alexandru Gutan]

Dear all,

I followed the instructions found here:
https://ci.apache.org/projects/flink/flink-docs-release-1.7/ops/deployment/kubernetes.html
Minikube version 0.31-01
Kubernetes version 1.10
Flink Docker image: flink:latest (1.7.0-scala_2.12)

I ran the following commands:

minikube start
minikube ssh 'sudo ip link set docker0 promisc on'
kubectl create -f jobmanager-deployment.yaml
kubectl create -f taskmanager-deployment.yaml
kubectl create -f jobmanager-service.yaml

The 2 taskmanagers fail.
Output:

Starting Task Manager
config file:
jobmanager.rpc.address: flink-jobmanager
jobmanager.rpc.port: 6123
jobmanager.heap.size: 1024m
taskmanager.heap.size: 1024m
taskmanager.numberOfTaskSlots: 2
parallelism.default: 1
rest.port: 8081
blob.server.port: 6124
query.server.port: 6125
Starting taskexecutor as a console application on host
flink-taskmanager-7679c9d55d-n2trk.
2018-12-19 11:42:45,216 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -

2018-12-19 11:42:45,218 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  Starting
TaskManager (Version: 1.7.0, Rev:49da9f9, Date:28.11.2018 @ 17:59:06 UTC)
2018-12-19 11:42:45,218 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  OS current
user: flink
2018-12-19 11:42:45,219 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  *Current
Hadoop/Kerberos user: *
2018-12-19 11:42:45,219 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  JVM:
OpenJDK 64-Bit Server VM - Oracle Corporation - 1.8/25.181-b13
2018-12-19 11:42:45,219 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  Maximum
heap size: 922 MiBytes
2018-12-19 11:42:45,220 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  JAVA_HOME:
/docker-java-home/jre
2018-12-19 11:42:45,220 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  No Hadoop
Dependency available
2018-12-19 11:42:45,221 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  JVM
Options:
2018-12-19 11:42:45,221 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -
-XX:+UseG1GC
2018-12-19 11:42:45,221 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   - -Xms922M
2018-12-19 11:42:45,221 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   - -Xmx922M
2018-12-19 11:42:45,221 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -
-XX:MaxDirectMemorySize=8388607T
2018-12-19 11:42:45,223 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -
-Dlog4j.configuration=file:/opt/flink/conf/log4j-console.properties
2018-12-19 11:42:45,223 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -
-Dlogback.configurationFile=file:/opt/flink/conf/logback-console.xml
2018-12-19 11:42:45,223 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  Program
Arguments:
2018-12-19 11:42:45,223 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -
--configDir
2018-12-19 11:42:45,224 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -
/opt/flink/conf
2018-12-19 11:42:45,224 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -  Classpath:
/opt/flink/lib/flink-python_2.12-1.7.0.jar:/opt/flink/lib/log4j-1.2.17.jar:/opt/flink/lib/slf4j-log4j12-1.7.15.jar:/opt/flink/lib/flink-dist_2.12-1.7.0.jar:::
2018-12-19 11:42:45,224 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   -

2018-12-19 11:42:45,228 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   - Registered
UNIX signal handlers for [TERM, HUP, INT]
2018-12-19 11:42:45,233 INFO
org.apache.flink.runtime.taskexecutor.TaskManagerRunner   - Maximum
number of open file descriptors is 1048576.
2018-12-19 11:42:45,249 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: jobmanager.rpc.address, flink-jobmanager
2018-12-19 11:42:45,250 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: jobmanager.rpc.port, 6123
2018-12-19 11:42:45,251 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: jobmanager.heap.size, 1024m
2018-12-19 11:42:45,251 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: taskmanager.heap.size, 1024m
2018-12-19 11:42:45,251 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: taskmanager.numberOfTaskSlots, 2
2018-12-19 11:42:45,252 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: parallelism.default, 1
2018-12-19 11:42:45,252 INFO
org.apache.flink.configuration.GlobalConfiguration- Loading
configuration property: 

Re: Is Flink:1.5 Docker image broken?

[Alexandru Gutan]

Well those are unofficial, so you might raise a correspoinding issue on
github (since the images are there) for that.

On 31 May 2018 at 08:09, Chirag Dewan  wrote:

> Hi,
>
> flink:latest docker image doesn't seem to work. I am not able to access
> the Flink Dashboard after deploying it on Kubernetes.
>
> Anyone else facing the issue?
>
> Thanks,
>
> Chirag
>

Re: Flink - Kafka Connector

[Alexandru Gutan]

You will be able to use it. Kafka 1.10 has backwards compatibility with
v1.0, 0.11 and 0.10 connectors as far as I know.

On 13 April 2018 at 15:12, Lehuede sebastien  wrote:

> Hi All,
>
> I'm very new in Flink (And on Streaming Application topic in general) so
> sorry if for my newbie question.
>
> I plan to do some test with Kafka and Flink and use the Kafka connector
> for that.
>
> I find information on this page : https://ci.apache.org/
> projects/flink/flink-docs-release-1.4/dev/connectors/kafka.html
>
> Can you confirm i will not be able to use latest version of Kafka (1.1.0),
> Flink 1.4 and this connector to do my test ? The connector is compatible
> with Kafka 0.11 and under only ?
>
> Regards,
> Sebastien.
>
>
>

Re: Table/SQL Kafka Sink Question

[Alexandru Gutan]

That's what I concluded as well after checking the docs and source code.

I'm thinking to add another job using the Stream API (where it is
possible), that will ingest the data resulted from by Table/SQL API job,
and that will add the message key into Kafka.

On 27 March 2018 at 12:55, Chesnay Schepler  wrote:

> Hello,
>
> as far as i can this is not possible. I'm including Timo, maybe he can
> explain why this isn't supported.
>
>
> On 26.03.2018 21:56, Pavel Ciorba wrote:
>
> Hi everyone!
>
> Can I specify a *message key* using the Kafka sink in the Table/SQL API ?
> The goal is to sink each row as JSON along side with a message key into
> Kafka.
>
> I was achieving it using the Stream API by specifying a
> *KeyedSerializationSchema* using the *serializeKey() *method.
>
> Thanks in advance!
>
>
>

Flink on DCOS on DigitalOcean

[Alexandru Gutan]

Dear all,

Can somebody confirm to have successfully installed Flink on DCOS (running
on DigitalOcean) ?

I've raised an issue on StackOverflow but nobody has assisted me.

I've described everything in detail here:

https://stackoverflow.com/questions/45391980/error-installing-flink-in-dcos

I deployed according to this instructions:
https://dcos.io/docs/1.9/installing/cloud/digitalocean/

Every other package from DCOS Universe seems to install without any problem.

Flink AggregateFunction Help

[Alexandru Gutan]

Hello!

I'm unable to see any results from .print() after performing an
AggregateFunction over a timeWindow.

I made a very detailed post on StackOverflow:
https://stackoverflow.com/questions/45898849/flink-aggregatefunction

Thank you!