You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Kaan Sancak <ka...@gmail.com> on 2020/04/22 00:58:01 UTC
Reading from sockets using dataset api
Hi,
I have been running some experiments on large graph data, smallest graph I have been using is around ~70 billion edges. I have a graph generator, which generates the graph in parallel and feeds to the running system. However, it takes a lot of time to read the edges, because even though the graph generation process is parallel, in Flink I can only listen from master node (correct me if I am wrong). Another option is dumping the generated data to a file and reading with readFromCsv, however this is not feasible in terms of storage management.
What I want to do is, invoking my graph generator, using ipc/tcp protocols and reading the generated data from the sockets. Since the graph data is also generated parallel in each node, I want to make use of ipc, and read the data in parallel at each node. I made some online digging but couldn’t find something similar using dataset api. I would be glad if you have some similar use cases or examples.
Is it possible to use streaming environment to create the data in parallel and switch to dataset api?
Thanks in advance!
Best
Kaan
Re: Reading from sockets using dataset api
Posted by Arvid Heise <ar...@ververica.com>.
Hi Kaan,
explicitly mapping to physical nodes is currently not supported and would
need some workarounds. I have readded user mailing list (please always also
include it in response); maybe someone can help you with that.
Best,
Arvid
On Thu, Apr 30, 2020 at 10:12 AM Kaan Sancak <ka...@gmail.com> wrote:
> One quick question tho, on each generator node I am opening 24 sockets
> (number of cores that I have). Is there a way to guarantee that while doing
> the map function, each of the slave nodes distributes this 24 socket ports
> between its task slots(each slave also have 24 slave),
> Sorry, I have asked a lot questions.
>
> Stay safe!
> Kaan
>
> On Thu, Apr 30, 2020 at 3:06 AM Kaan Sancak <ka...@gmail.com> wrote:
>
>> Hi Arvid,
>> As you said, I am only interested in batch processing right now. And it
>> seems to be working fine now.
>>
>> Thanks for your help!
>> Best
>> Kaan
>>
>> On Thu, Apr 30, 2020 at 2:31 AM Arvid Heise <ar...@ververica.com> wrote:
>>
>>> Hi Kaan,
>>>
>>> not entirely sure I understand your solution. I gathered that you create
>>> a dataset of TCP addresses and then use flatMap to fetch and output the
>>> data?
>>>
>>> If so, then I think it's a good solution for batch processing (DataSet).
>>> It doesn't work in DataStream because it doesn't play well with
>>> checkpointing, but you seem to be interested only in batch. It's also not
>>> the first time, I have seen this pattern being used in batch.
>>>
>>> In general, if it works and is fast enough, it's always a good solution
>>> ;). No need to make it more complicated if you can solve it with simpler
>>> means and you can maintain it more easily.
>>>
>>> Best,
>>>
>>> Arvid
>>>
>>> On Wed, Apr 29, 2020 at 10:19 PM Kaan Sancak <ka...@gmail.com> wrote:
>>>
>>>> Hi Arvid,
>>>>
>>>> I have implemented a zmq listener class without extending any class of
>>>> Flink.
>>>> The listener has a constructor with the port number.
>>>>
>>>> Then in the execution I have created a dateset of string which has the
>>>> port numbers.
>>>> Then I used a flattop function, which returned Tuple2<Long, Long>. I
>>>> opened the tcp sockets using localhost, so matching was quite easy.
>>>>
>>>> This seemed to work for me. What do you think about this
>>>> implementation. Do you see any drawback?
>>>>
>>>> Best
>>>> Kaan
>>>>
>>>> On Apr 29, 2020, at 7:40 AM, Arvid Heise <ar...@ververica.com> wrote:
>>>>
>>>> Hi Kaan,
>>>>
>>>> seems like ZMQ is using TCP and not HTTP. So I guess the easiest way
>>>> would be to use a ZMQ Java binding to access it [1].
>>>>
>>>> But of course, it's much more complicated to write an iterator logic
>>>> for that. Not sure how ZMQ signals the end of such a graph? Maybe it closes
>>>> the socket and you can just read as much as possible.
>>>>
>>>> [1] https://zeromq.org/languages/java/
>>>>
>>>> On Tue, Apr 28, 2020 at 10:56 PM Kaan Sancak <ka...@gmail.com>
>>>> wrote:
>>>>
>>>>> Hi Arvid,
>>>>>
>>>>> I am sorry for the late response. I had some deadlines, but I am back
>>>>> to work now.
>>>>> I have been trying to implement what we have talked. But I am having
>>>>> problems on the implementation.
>>>>> I have been using ZMQ to open sockets, because that is inheritenly
>>>>> supported in my graph generator. But, I couldn’t make the connection using
>>>>> input streams.
>>>>> Do you have any specific examples, where I can look and have a better
>>>>> idea on how to implement this?
>>>>>
>>>>> Best
>>>>> Kaan
>>>>>
>>>>> On Apr 24, 2020, at 4:37 AM, Arvid Heise <ar...@ververica.com> wrote:
>>>>>
>>>>> Hm, I confused sockets to work the other way around (so pulling like
>>>>> URLInputStream instead of listening). I'd go by providing the data on
>>>>> a port on each generator node. And then read from that in multiple sources.
>>>>>
>>>>> I think the best solution is to implement a custom InputFormat and
>>>>> then use readInput. You could implement a subclass of
>>>>> GenericInputFormat. You might even use IteratorInputFormat like this:
>>>>>
>>>>> private static class URLInputIterator implements Iterator<Tuple2<Long, Long>>, Serializable {
>>>>> private final URL url;
>>>>> private Iterator<Tuple2<Long, Long>> inner;
>>>>>
>>>>> private URLInputIterator(URL url) {
>>>>> this.url = url;
>>>>> }
>>>>>
>>>>> private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
>>>>> InputStream inputStream = url.openStream();
>>>>> inner = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8))
>>>>> .lines()
>>>>> .map(line -> {
>>>>> String[] parts = line.split(";");
>>>>> return new Tuple2<>(Long.parseLong(parts[0]), Long.parseLong(parts[1]));
>>>>> })
>>>>> .iterator();
>>>>> }
>>>>>
>>>>> @Override
>>>>> public boolean hasNext() {
>>>>> return inner.hasNext();
>>>>> }
>>>>>
>>>>> @Override
>>>>> public Tuple2<Long, Long> next() {
>>>>> return inner.next();
>>>>> }
>>>>> }
>>>>>
>>>>> env.fromCollection(new URLInputIterator(new URL("gen_node1", 9999)), Types.TUPLE(Types.LONG, Types.LONG));
>>>>>
>>>>>
>>>>>
>>>>>
>>>>> On Fri, Apr 24, 2020 at 9:42 AM Kaan Sancak <ka...@gmail.com>
>>>>> wrote:
>>>>>
>>>>>> Yes, that sounds like a great idea and actually that's what I am
>>>>>> trying to do.
>>>>>>
>>>>>> Then you configure your analysis job to read from each of these
>>>>>> sockets with a separate source and union them before feeding them to the
>>>>>> actual job?
>>>>>>
>>>>>>
>>>>>> Before trying to open the sockets on the slave nodes, first I have
>>>>>> opened just one socket at master node, and I also run the generator with
>>>>>> one node as well. I was able to read the graph, and the run my algorithm
>>>>>> without any problems. This was a test run to see whatever I can do it.
>>>>>>
>>>>>> After, I have opened bunch of sockets on my generators, now I am
>>>>>> trying to configure Flink to read from those sockets. However, I am having
>>>>>> problems while trying to assign each task manager to a separate socket. I
>>>>>> am assuming my problems are related to network binding. In my configuration
>>>>>> file, jobmanager.rpc.address is set but I have not done
>>>>>> similar configurations for slave nodes.
>>>>>>
>>>>>> Am I on the right track, or is there an easier way to handle this?
>>>>>>
>>>>>> I think my point is how to do `read from each of these sockets with
>>>>>> a separate source` part.
>>>>>>
>>>>>> Thanks again
>>>>>>
>>>>>> Best
>>>>>> Kaan
>>>>>>
>>>>>>
>>>>>>
>>>>>> On Apr 24, 2020, at 3:11 AM, Arvid Heise <ar...@ververica.com> wrote:
>>>>>>
>>>>>> Hi Kaan,
>>>>>>
>>>>>> sorry, I haven't considered I/O as the bottleneck. I thought a bit
>>>>>> more about your issue and came to a rather simple solution.
>>>>>>
>>>>>> How about you open a socket on each of your generator nodes? Then you
>>>>>> configure your analysis job to read from each of these sockets with a
>>>>>> separate source and union them before feeding them to the actual job?
>>>>>>
>>>>>> You don't need to modify much on the analysis job and each source can
>>>>>> be independently read. WDYT?
>>>>>>
>>>>>> On Fri, Apr 24, 2020 at 8:46 AM Kaan Sancak <ka...@gmail.com>
>>>>>> wrote:
>>>>>>
>>>>>>> Thanks for the answer! Also thanks for raising some concerns about
>>>>>>> my question.
>>>>>>>
>>>>>>> Some of the graphs I have been using is larger than 1.5 tb, and I am
>>>>>>> currently an experiment stage of a project, and I am making modifications
>>>>>>> to my code and re-runing the experiments again. Currently, on some of the
>>>>>>> largest graphs I have been using, IO became an issue for me and keeps me
>>>>>>> wait for couple of hours.
>>>>>>>
>>>>>>> Moreover, I have a parallel/distributed graph generator, which I can
>>>>>>> run on the same set of nodes in my cluster. So what I wanted to do was, to
>>>>>>> run my Flink program and graph generator at the same time and feed the
>>>>>>> graph through generator, which should be faster than making IO from the
>>>>>>> disk. As you said, it is not essential for me to that, but I am trying to
>>>>>>> see what I am able to do using Flink and how can I solve such problems. I
>>>>>>> was also using another framework, and faced with the similar problem, I was
>>>>>>> able to reduce the graph read time from hours to minutes using this method.
>>>>>>>
>>>>>>> Do you really have more main memory than disk space?
>>>>>>>
>>>>>>>
>>>>>>> My issue is actually not storage related, I am trying to see how can
>>>>>>> I reduce the IO time.
>>>>>>>
>>>>>>> One trick came to my mind is, creating dummy dataset, and using a
>>>>>>> map function on the dataset, I can open-up bunch of sockets and listen the
>>>>>>> generator, and collect the generated data. I am trying to see how it will
>>>>>>> turn out.
>>>>>>>
>>>>>>> Alternatively, if graph generation is rather cheap, you could also
>>>>>>> try to incorporate it directly into the analysis job.
>>>>>>>
>>>>>>>
>>>>>>> I am not familiar with the analysis jobs. I will look into it.
>>>>>>>
>>>>>>> Again, this is actually not a problem, I am just trying to
>>>>>>> experiment with the framework and see what I can do. I am very new to
>>>>>>> Flink, so my methods might be wrong. Thanks for the help!
>>>>>>>
>>>>>>> Best
>>>>>>> Kaan
>>>>>>>
>>>>>>>
>>>>>>> On Apr 23, 2020, at 10:51 AM, Arvid Heise <ar...@ververica.com>
>>>>>>> wrote:
>>>>>>>
>>>>>>> Hi Kaan,
>>>>>>>
>>>>>>> afaik there is no (easy) way to switch from streaming back to batch
>>>>>>> API while retaining all data in memory (correct me if I misunderstood).
>>>>>>>
>>>>>>> However, from your description, I also have some severe
>>>>>>> understanding problems. Why can't you dump the data to some file? Do you
>>>>>>> really have more main memory than disk space? Or do you have no shared
>>>>>>> memory between your generating cluster and the flink cluster?
>>>>>>>
>>>>>>> It almost sounds as if the issue at heart is rather to find a good
>>>>>>> serialization format on how to store the edges. The 70 billion edges could
>>>>>>> be stored in an array of id pairs, which amount to ~560 GB uncompressed
>>>>>>> data if stored in Avro (or any other binary serialization format) when ids
>>>>>>> are longs. That's not much by today's standards and could also be easily
>>>>>>> offloaded to S3.
>>>>>>>
>>>>>>> Alternatively, if graph generation is rather cheap, you could also
>>>>>>> try to incorporate it directly into the analysis job.
>>>>>>>
>>>>>>> On Wed, Apr 22, 2020 at 2:58 AM Kaan Sancak <ka...@gmail.com>
>>>>>>> wrote:
>>>>>>>
>>>>>>>> Hi,
>>>>>>>>
>>>>>>>> I have been running some experiments on large graph data, smallest
>>>>>>>> graph I have been using is around ~70 billion edges. I have a graph
>>>>>>>> generator, which generates the graph in parallel and feeds to the running
>>>>>>>> system. However, it takes a lot of time to read the edges, because even
>>>>>>>> though the graph generation process is parallel, in Flink I can only listen
>>>>>>>> from master node (correct me if I am wrong). Another option is dumping the
>>>>>>>> generated data to a file and reading with readFromCsv, however this is not
>>>>>>>> feasible in terms of storage management.
>>>>>>>>
>>>>>>>> What I want to do is, invoking my graph generator, using ipc/tcp
>>>>>>>> protocols and reading the generated data from the sockets. Since the graph
>>>>>>>> data is also generated parallel in each node, I want to make use of ipc,
>>>>>>>> and read the data in parallel at each node. I made some online digging but
>>>>>>>> couldn’t find something similar using dataset api. I would be glad if you
>>>>>>>> have some similar use cases or examples.
>>>>>>>>
>>>>>>>> Is it possible to use streaming environment to create the data in
>>>>>>>> parallel and switch to dataset api?
>>>>>>>>
>>>>>>>> Thanks in advance!
>>>>>>>>
>>>>>>>> Best
>>>>>>>> Kaan
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>> --
>>>>>>> Arvid Heise | Senior Java Developer
>>>>>>> <https://www.ververica.com/>
>>>>>>>
>>>>>>> Follow us @VervericaData
>>>>>>> --
>>>>>>> Join Flink Forward <https://flink-forward.org/> - The Apache Flink
>>>>>>> Conference
>>>>>>> Stream Processing | Event Driven | Real Time
>>>>>>> --
>>>>>>> Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
>>>>>>> --
>>>>>>> Ververica GmbH
>>>>>>> Registered at Amtsgericht Charlottenburg: HRB 158244 B
>>>>>>> Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason,
>>>>>>> Ji (Toni) Cheng
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>
>>>>>> --
>>>>>> Arvid Heise | Senior Java Developer
>>>>>> <https://www.ververica.com/>
>>>>>>
>>>>>> Follow us @VervericaData
>>>>>> --
>>>>>> Join Flink Forward <https://flink-forward.org/> - The Apache Flink
>>>>>> Conference
>>>>>> Stream Processing | Event Driven | Real Time
>>>>>> --
>>>>>> Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
>>>>>> --
>>>>>> Ververica GmbH
>>>>>> Registered at Amtsgericht Charlottenburg: HRB 158244 B
>>>>>> Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason,
>>>>>> Ji (Toni) Cheng
>>>>>>
>>>>>>
>>>>>>
>>>>>
>>>>> --
>>>>> Arvid Heise | Senior Java Developer
>>>>> <https://www.ververica.com/>
>>>>>
>>>>> Follow us @VervericaData
>>>>> --
>>>>> Join Flink Forward <https://flink-forward.org/> - The Apache Flink
>>>>> Conference
>>>>> Stream Processing | Event Driven | Real Time
>>>>> --
>>>>> Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
>>>>> --
>>>>> Ververica GmbH
>>>>> Registered at Amtsgericht Charlottenburg: HRB 158244 B
>>>>> Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason,
>>>>> Ji (Toni) Cheng
>>>>>
>>>>>
>>>>>
>>>>
>>>> --
>>>> Arvid Heise | Senior Java Developer
>>>> <https://www.ververica.com/>
>>>>
>>>> Follow us @VervericaData
>>>> --
>>>> Join Flink Forward <https://flink-forward.org/> - The Apache Flink
>>>> Conference
>>>> Stream Processing | Event Driven | Real Time
>>>> --
>>>> Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
>>>> --
>>>> Ververica GmbH
>>>> Registered at Amtsgericht Charlottenburg: HRB 158244 B
>>>> Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji
>>>> (Toni) Cheng
>>>>
>>>>
>>>>
>>>
>>> --
>>>
>>> Arvid Heise | Senior Java Developer
>>>
>>> <https://www.ververica.com/>
>>>
>>> Follow us @VervericaData
>>>
>>> --
>>>
>>> Join Flink Forward <https://flink-forward.org/> - The Apache Flink
>>> Conference
>>>
>>> Stream Processing | Event Driven | Real Time
>>>
>>> --
>>>
>>> Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
>>>
>>> --
>>> Ververica GmbH
>>> Registered at Amtsgericht Charlottenburg: HRB 158244 B
>>> Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji
>>> (Toni) Cheng
>>>
>>
--
Arvid Heise | Senior Java Developer
<https://www.ververica.com/>
Follow us @VervericaData
--
Join Flink Forward <https://flink-forward.org/> - The Apache Flink
Conference
Stream Processing | Event Driven | Real Time
--
Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
--
Ververica GmbH
Registered at Amtsgericht Charlottenburg: HRB 158244 B
Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji
(Toni) Cheng
Re: Reading from sockets using dataset api
Posted by Arvid Heise <ar...@ververica.com>.
Hi Kaan,
not entirely sure I understand your solution. I gathered that you create a
dataset of TCP addresses and then use flatMap to fetch and output the data?
If so, then I think it's a good solution for batch processing (DataSet). It
doesn't work in DataStream because it doesn't play well with checkpointing,
but you seem to be interested only in batch. It's also not the first time,
I have seen this pattern being used in batch.
In general, if it works and is fast enough, it's always a good solution ;).
No need to make it more complicated if you can solve it with simpler means
and you can maintain it more easily.
Best,
Arvid
On Wed, Apr 29, 2020 at 10:19 PM Kaan Sancak <ka...@gmail.com> wrote:
> Hi Arvid,
>
> I have implemented a zmq listener class without extending any class of
> Flink.
> The listener has a constructor with the port number.
>
> Then in the execution I have created a dateset of string which has the
> port numbers.
> Then I used a flattop function, which returned Tuple2<Long, Long>. I
> opened the tcp sockets using localhost, so matching was quite easy.
>
> This seemed to work for me. What do you think about this implementation.
> Do you see any drawback?
>
> Best
> Kaan
>
> On Apr 29, 2020, at 7:40 AM, Arvid Heise <ar...@ververica.com> wrote:
>
> Hi Kaan,
>
> seems like ZMQ is using TCP and not HTTP. So I guess the easiest way would
> be to use a ZMQ Java binding to access it [1].
>
> But of course, it's much more complicated to write an iterator logic for
> that. Not sure how ZMQ signals the end of such a graph? Maybe it closes the
> socket and you can just read as much as possible.
>
> [1] https://zeromq.org/languages/java/
>
> On Tue, Apr 28, 2020 at 10:56 PM Kaan Sancak <ka...@gmail.com> wrote:
>
>> Hi Arvid,
>>
>> I am sorry for the late response. I had some deadlines, but I am back to
>> work now.
>> I have been trying to implement what we have talked. But I am having
>> problems on the implementation.
>> I have been using ZMQ to open sockets, because that is inheritenly
>> supported in my graph generator. But, I couldn’t make the connection using
>> input streams.
>> Do you have any specific examples, where I can look and have a better
>> idea on how to implement this?
>>
>> Best
>> Kaan
>>
>> On Apr 24, 2020, at 4:37 AM, Arvid Heise <ar...@ververica.com> wrote:
>>
>> Hm, I confused sockets to work the other way around (so pulling like
>> URLInputStream instead of listening). I'd go by providing the data on a
>> port on each generator node. And then read from that in multiple sources.
>>
>> I think the best solution is to implement a custom InputFormat and then
>> use readInput. You could implement a subclass of GenericInputFormat. You
>> might even use IteratorInputFormat like this:
>>
>> private static class URLInputIterator implements Iterator<Tuple2<Long, Long>>, Serializable {
>> private final URL url;
>> private Iterator<Tuple2<Long, Long>> inner;
>>
>> private URLInputIterator(URL url) {
>> this.url = url;
>> }
>>
>> private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
>> InputStream inputStream = url.openStream();
>> inner = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8))
>> .lines()
>> .map(line -> {
>> String[] parts = line.split(";");
>> return new Tuple2<>(Long.parseLong(parts[0]), Long.parseLong(parts[1]));
>> })
>> .iterator();
>> }
>>
>> @Override
>> public boolean hasNext() {
>> return inner.hasNext();
>> }
>>
>> @Override
>> public Tuple2<Long, Long> next() {
>> return inner.next();
>> }
>> }
>>
>> env.fromCollection(new URLInputIterator(new URL("gen_node1", 9999)), Types.TUPLE(Types.LONG, Types.LONG));
>>
>>
>>
>>
>> On Fri, Apr 24, 2020 at 9:42 AM Kaan Sancak <ka...@gmail.com> wrote:
>>
>>> Yes, that sounds like a great idea and actually that's what I am trying
>>> to do.
>>>
>>> Then you configure your analysis job to read from each of these sockets
>>> with a separate source and union them before feeding them to the actual job?
>>>
>>>
>>> Before trying to open the sockets on the slave nodes, first I have
>>> opened just one socket at master node, and I also run the generator with
>>> one node as well. I was able to read the graph, and the run my algorithm
>>> without any problems. This was a test run to see whatever I can do it.
>>>
>>> After, I have opened bunch of sockets on my generators, now I am trying
>>> to configure Flink to read from those sockets. However, I am having
>>> problems while trying to assign each task manager to a separate socket. I
>>> am assuming my problems are related to network binding. In my configuration
>>> file, jobmanager.rpc.address is set but I have not done
>>> similar configurations for slave nodes.
>>>
>>> Am I on the right track, or is there an easier way to handle this?
>>>
>>> I think my point is how to do `read from each of these sockets with a
>>> separate source` part.
>>>
>>> Thanks again
>>>
>>> Best
>>> Kaan
>>>
>>>
>>>
>>> On Apr 24, 2020, at 3:11 AM, Arvid Heise <ar...@ververica.com> wrote:
>>>
>>> Hi Kaan,
>>>
>>> sorry, I haven't considered I/O as the bottleneck. I thought a bit more
>>> about your issue and came to a rather simple solution.
>>>
>>> How about you open a socket on each of your generator nodes? Then you
>>> configure your analysis job to read from each of these sockets with a
>>> separate source and union them before feeding them to the actual job?
>>>
>>> You don't need to modify much on the analysis job and each source can be
>>> independently read. WDYT?
>>>
>>> On Fri, Apr 24, 2020 at 8:46 AM Kaan Sancak <ka...@gmail.com> wrote:
>>>
>>>> Thanks for the answer! Also thanks for raising some concerns about my
>>>> question.
>>>>
>>>> Some of the graphs I have been using is larger than 1.5 tb, and I am
>>>> currently an experiment stage of a project, and I am making modifications
>>>> to my code and re-runing the experiments again. Currently, on some of the
>>>> largest graphs I have been using, IO became an issue for me and keeps me
>>>> wait for couple of hours.
>>>>
>>>> Moreover, I have a parallel/distributed graph generator, which I can
>>>> run on the same set of nodes in my cluster. So what I wanted to do was, to
>>>> run my Flink program and graph generator at the same time and feed the
>>>> graph through generator, which should be faster than making IO from the
>>>> disk. As you said, it is not essential for me to that, but I am trying to
>>>> see what I am able to do using Flink and how can I solve such problems. I
>>>> was also using another framework, and faced with the similar problem, I was
>>>> able to reduce the graph read time from hours to minutes using this method.
>>>>
>>>> Do you really have more main memory than disk space?
>>>>
>>>>
>>>> My issue is actually not storage related, I am trying to see how can I
>>>> reduce the IO time.
>>>>
>>>> One trick came to my mind is, creating dummy dataset, and using a map
>>>> function on the dataset, I can open-up bunch of sockets and listen the
>>>> generator, and collect the generated data. I am trying to see how it will
>>>> turn out.
>>>>
>>>> Alternatively, if graph generation is rather cheap, you could also try
>>>> to incorporate it directly into the analysis job.
>>>>
>>>>
>>>> I am not familiar with the analysis jobs. I will look into it.
>>>>
>>>> Again, this is actually not a problem, I am just trying to experiment
>>>> with the framework and see what I can do. I am very new to Flink, so my
>>>> methods might be wrong. Thanks for the help!
>>>>
>>>> Best
>>>> Kaan
>>>>
>>>>
>>>> On Apr 23, 2020, at 10:51 AM, Arvid Heise <ar...@ververica.com> wrote:
>>>>
>>>> Hi Kaan,
>>>>
>>>> afaik there is no (easy) way to switch from streaming back to batch API
>>>> while retaining all data in memory (correct me if I misunderstood).
>>>>
>>>> However, from your description, I also have some severe understanding
>>>> problems. Why can't you dump the data to some file? Do you really have more
>>>> main memory than disk space? Or do you have no shared memory between your
>>>> generating cluster and the flink cluster?
>>>>
>>>> It almost sounds as if the issue at heart is rather to find a good
>>>> serialization format on how to store the edges. The 70 billion edges could
>>>> be stored in an array of id pairs, which amount to ~560 GB uncompressed
>>>> data if stored in Avro (or any other binary serialization format) when ids
>>>> are longs. That's not much by today's standards and could also be easily
>>>> offloaded to S3.
>>>>
>>>> Alternatively, if graph generation is rather cheap, you could also try
>>>> to incorporate it directly into the analysis job.
>>>>
>>>> On Wed, Apr 22, 2020 at 2:58 AM Kaan Sancak <ka...@gmail.com> wrote:
>>>>
>>>>> Hi,
>>>>>
>>>>> I have been running some experiments on large graph data, smallest
>>>>> graph I have been using is around ~70 billion edges. I have a graph
>>>>> generator, which generates the graph in parallel and feeds to the running
>>>>> system. However, it takes a lot of time to read the edges, because even
>>>>> though the graph generation process is parallel, in Flink I can only listen
>>>>> from master node (correct me if I am wrong). Another option is dumping the
>>>>> generated data to a file and reading with readFromCsv, however this is not
>>>>> feasible in terms of storage management.
>>>>>
>>>>> What I want to do is, invoking my graph generator, using ipc/tcp
>>>>> protocols and reading the generated data from the sockets. Since the graph
>>>>> data is also generated parallel in each node, I want to make use of ipc,
>>>>> and read the data in parallel at each node. I made some online digging but
>>>>> couldn’t find something similar using dataset api. I would be glad if you
>>>>> have some similar use cases or examples.
>>>>>
>>>>> Is it possible to use streaming environment to create the data in
>>>>> parallel and switch to dataset api?
>>>>>
>>>>> Thanks in advance!
>>>>>
>>>>> Best
>>>>> Kaan
>>>>
>>>>
>>>>
>>>> --
>>>> Arvid Heise | Senior Java Developer
>>>> <https://www.ververica.com/>
>>>>
>>>> Follow us @VervericaData
>>>> --
>>>> Join Flink Forward <https://flink-forward.org/> - The Apache Flink
>>>> Conference
>>>> Stream Processing | Event Driven | Real Time
>>>> --
>>>> Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
>>>> --
>>>> Ververica GmbH
>>>> Registered at Amtsgericht Charlottenburg: HRB 158244 B
>>>> Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji
>>>> (Toni) Cheng
>>>>
>>>>
>>>>
>>>
>>> --
>>> Arvid Heise | Senior Java Developer
>>> <https://www.ververica.com/>
>>>
>>> Follow us @VervericaData
>>> --
>>> Join Flink Forward <https://flink-forward.org/> - The Apache Flink
>>> Conference
>>> Stream Processing | Event Driven | Real Time
>>> --
>>> Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
>>> --
>>> Ververica GmbH
>>> Registered at Amtsgericht Charlottenburg: HRB 158244 B
>>> Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji
>>> (Toni) Cheng
>>>
>>>
>>>
>>
>> --
>> Arvid Heise | Senior Java Developer
>> <https://www.ververica.com/>
>>
>> Follow us @VervericaData
>> --
>> Join Flink Forward <https://flink-forward.org/> - The Apache Flink
>> Conference
>> Stream Processing | Event Driven | Real Time
>> --
>> Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
>> --
>> Ververica GmbH
>> Registered at Amtsgericht Charlottenburg: HRB 158244 B
>> Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji
>> (Toni) Cheng
>>
>>
>>
>
> --
> Arvid Heise | Senior Java Developer
> <https://www.ververica.com/>
>
> Follow us @VervericaData
> --
> Join Flink Forward <https://flink-forward.org/> - The Apache Flink
> Conference
> Stream Processing | Event Driven | Real Time
> --
> Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
> --
> Ververica GmbH
> Registered at Amtsgericht Charlottenburg: HRB 158244 B
> Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji
> (Toni) Cheng
>
>
>
--
Arvid Heise | Senior Java Developer
<https://www.ververica.com/>
Follow us @VervericaData
--
Join Flink Forward <https://flink-forward.org/> - The Apache Flink
Conference
Stream Processing | Event Driven | Real Time
--
Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
--
Ververica GmbH
Registered at Amtsgericht Charlottenburg: HRB 158244 B
Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji
(Toni) Cheng
Re: Reading from sockets using dataset api
Posted by Arvid Heise <ar...@ververica.com>.
Hi Kaan,
seems like ZMQ is using TCP and not HTTP. So I guess the easiest way would
be to use a ZMQ Java binding to access it [1].
But of course, it's much more complicated to write an iterator logic for
that. Not sure how ZMQ signals the end of such a graph? Maybe it closes the
socket and you can just read as much as possible.
[1] https://zeromq.org/languages/java/
On Tue, Apr 28, 2020 at 10:56 PM Kaan Sancak <ka...@gmail.com> wrote:
> Hi Arvid,
>
> I am sorry for the late response. I had some deadlines, but I am back to
> work now.
> I have been trying to implement what we have talked. But I am having
> problems on the implementation.
> I have been using ZMQ to open sockets, because that is inheritenly
> supported in my graph generator. But, I couldn’t make the connection using
> input streams.
> Do you have any specific examples, where I can look and have a better idea
> on how to implement this?
>
> Best
> Kaan
>
> On Apr 24, 2020, at 4:37 AM, Arvid Heise <ar...@ververica.com> wrote:
>
> Hm, I confused sockets to work the other way around (so pulling like
> URLInputStream instead of listening). I'd go by providing the data on a
> port on each generator node. And then read from that in multiple sources.
>
> I think the best solution is to implement a custom InputFormat and then
> use readInput. You could implement a subclass of GenericInputFormat. You
> might even use IteratorInputFormat like this:
>
> private static class URLInputIterator implements Iterator<Tuple2<Long, Long>>, Serializable {
> private final URL url;
> private Iterator<Tuple2<Long, Long>> inner;
>
> private URLInputIterator(URL url) {
> this.url = url;
> }
>
> private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
> InputStream inputStream = url.openStream();
> inner = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8))
> .lines()
> .map(line -> {
> String[] parts = line.split(";");
> return new Tuple2<>(Long.parseLong(parts[0]), Long.parseLong(parts[1]));
> })
> .iterator();
> }
>
> @Override
> public boolean hasNext() {
> return inner.hasNext();
> }
>
> @Override
> public Tuple2<Long, Long> next() {
> return inner.next();
> }
> }
>
> env.fromCollection(new URLInputIterator(new URL("gen_node1", 9999)), Types.TUPLE(Types.LONG, Types.LONG));
>
>
>
>
> On Fri, Apr 24, 2020 at 9:42 AM Kaan Sancak <ka...@gmail.com> wrote:
>
>> Yes, that sounds like a great idea and actually that's what I am trying
>> to do.
>>
>> Then you configure your analysis job to read from each of these sockets
>> with a separate source and union them before feeding them to the actual job?
>>
>>
>> Before trying to open the sockets on the slave nodes, first I have opened
>> just one socket at master node, and I also run the generator with one node
>> as well. I was able to read the graph, and the run my algorithm without any
>> problems. This was a test run to see whatever I can do it.
>>
>> After, I have opened bunch of sockets on my generators, now I am trying
>> to configure Flink to read from those sockets. However, I am having
>> problems while trying to assign each task manager to a separate socket. I
>> am assuming my problems are related to network binding. In my configuration
>> file, jobmanager.rpc.address is set but I have not done
>> similar configurations for slave nodes.
>>
>> Am I on the right track, or is there an easier way to handle this?
>>
>> I think my point is how to do `read from each of these sockets with a
>> separate source` part.
>>
>> Thanks again
>>
>> Best
>> Kaan
>>
>>
>>
>> On Apr 24, 2020, at 3:11 AM, Arvid Heise <ar...@ververica.com> wrote:
>>
>> Hi Kaan,
>>
>> sorry, I haven't considered I/O as the bottleneck. I thought a bit more
>> about your issue and came to a rather simple solution.
>>
>> How about you open a socket on each of your generator nodes? Then you
>> configure your analysis job to read from each of these sockets with a
>> separate source and union them before feeding them to the actual job?
>>
>> You don't need to modify much on the analysis job and each source can be
>> independently read. WDYT?
>>
>> On Fri, Apr 24, 2020 at 8:46 AM Kaan Sancak <ka...@gmail.com> wrote:
>>
>>> Thanks for the answer! Also thanks for raising some concerns about my
>>> question.
>>>
>>> Some of the graphs I have been using is larger than 1.5 tb, and I am
>>> currently an experiment stage of a project, and I am making modifications
>>> to my code and re-runing the experiments again. Currently, on some of the
>>> largest graphs I have been using, IO became an issue for me and keeps me
>>> wait for couple of hours.
>>>
>>> Moreover, I have a parallel/distributed graph generator, which I can run
>>> on the same set of nodes in my cluster. So what I wanted to do was, to run
>>> my Flink program and graph generator at the same time and feed the graph
>>> through generator, which should be faster than making IO from the disk. As
>>> you said, it is not essential for me to that, but I am trying to see what I
>>> am able to do using Flink and how can I solve such problems. I was also
>>> using another framework, and faced with the similar problem, I was able to
>>> reduce the graph read time from hours to minutes using this method.
>>>
>>> Do you really have more main memory than disk space?
>>>
>>>
>>> My issue is actually not storage related, I am trying to see how can I
>>> reduce the IO time.
>>>
>>> One trick came to my mind is, creating dummy dataset, and using a map
>>> function on the dataset, I can open-up bunch of sockets and listen the
>>> generator, and collect the generated data. I am trying to see how it will
>>> turn out.
>>>
>>> Alternatively, if graph generation is rather cheap, you could also try
>>> to incorporate it directly into the analysis job.
>>>
>>>
>>> I am not familiar with the analysis jobs. I will look into it.
>>>
>>> Again, this is actually not a problem, I am just trying to experiment
>>> with the framework and see what I can do. I am very new to Flink, so my
>>> methods might be wrong. Thanks for the help!
>>>
>>> Best
>>> Kaan
>>>
>>>
>>> On Apr 23, 2020, at 10:51 AM, Arvid Heise <ar...@ververica.com> wrote:
>>>
>>> Hi Kaan,
>>>
>>> afaik there is no (easy) way to switch from streaming back to batch API
>>> while retaining all data in memory (correct me if I misunderstood).
>>>
>>> However, from your description, I also have some severe understanding
>>> problems. Why can't you dump the data to some file? Do you really have more
>>> main memory than disk space? Or do you have no shared memory between your
>>> generating cluster and the flink cluster?
>>>
>>> It almost sounds as if the issue at heart is rather to find a good
>>> serialization format on how to store the edges. The 70 billion edges could
>>> be stored in an array of id pairs, which amount to ~560 GB uncompressed
>>> data if stored in Avro (or any other binary serialization format) when ids
>>> are longs. That's not much by today's standards and could also be easily
>>> offloaded to S3.
>>>
>>> Alternatively, if graph generation is rather cheap, you could also try
>>> to incorporate it directly into the analysis job.
>>>
>>> On Wed, Apr 22, 2020 at 2:58 AM Kaan Sancak <ka...@gmail.com> wrote:
>>>
>>>> Hi,
>>>>
>>>> I have been running some experiments on large graph data, smallest
>>>> graph I have been using is around ~70 billion edges. I have a graph
>>>> generator, which generates the graph in parallel and feeds to the running
>>>> system. However, it takes a lot of time to read the edges, because even
>>>> though the graph generation process is parallel, in Flink I can only listen
>>>> from master node (correct me if I am wrong). Another option is dumping the
>>>> generated data to a file and reading with readFromCsv, however this is not
>>>> feasible in terms of storage management.
>>>>
>>>> What I want to do is, invoking my graph generator, using ipc/tcp
>>>> protocols and reading the generated data from the sockets. Since the graph
>>>> data is also generated parallel in each node, I want to make use of ipc,
>>>> and read the data in parallel at each node. I made some online digging but
>>>> couldn’t find something similar using dataset api. I would be glad if you
>>>> have some similar use cases or examples.
>>>>
>>>> Is it possible to use streaming environment to create the data in
>>>> parallel and switch to dataset api?
>>>>
>>>> Thanks in advance!
>>>>
>>>> Best
>>>> Kaan
>>>
>>>
>>>
>>> --
>>> Arvid Heise | Senior Java Developer
>>> <https://www.ververica.com/>
>>>
>>> Follow us @VervericaData
>>> --
>>> Join Flink Forward <https://flink-forward.org/> - The Apache Flink
>>> Conference
>>> Stream Processing | Event Driven | Real Time
>>> --
>>> Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
>>> --
>>> Ververica GmbH
>>> Registered at Amtsgericht Charlottenburg: HRB 158244 B
>>> Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji
>>> (Toni) Cheng
>>>
>>>
>>>
>>
>> --
>> Arvid Heise | Senior Java Developer
>> <https://www.ververica.com/>
>>
>> Follow us @VervericaData
>> --
>> Join Flink Forward <https://flink-forward.org/> - The Apache Flink
>> Conference
>> Stream Processing | Event Driven | Real Time
>> --
>> Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
>> --
>> Ververica GmbH
>> Registered at Amtsgericht Charlottenburg: HRB 158244 B
>> Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji
>> (Toni) Cheng
>>
>>
>>
>
> --
> Arvid Heise | Senior Java Developer
> <https://www.ververica.com/>
>
> Follow us @VervericaData
> --
> Join Flink Forward <https://flink-forward.org/> - The Apache Flink
> Conference
> Stream Processing | Event Driven | Real Time
> --
> Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
> --
> Ververica GmbH
> Registered at Amtsgericht Charlottenburg: HRB 158244 B
> Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji
> (Toni) Cheng
>
>
>
--
Arvid Heise | Senior Java Developer
<https://www.ververica.com/>
Follow us @VervericaData
--
Join Flink Forward <https://flink-forward.org/> - The Apache Flink
Conference
Stream Processing | Event Driven | Real Time
--
Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
--
Ververica GmbH
Registered at Amtsgericht Charlottenburg: HRB 158244 B
Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji
(Toni) Cheng
Re: Reading from sockets using dataset api
Posted by Arvid Heise <ar...@ververica.com>.
Hm, I confused sockets to work the other way around (so pulling like
URLInputStream instead of listening). I'd go by providing the data on a
port on each generator node. And then read from that in multiple sources.
I think the best solution is to implement a custom InputFormat and then use
readInput. You could implement a subclass of GenericInputFormat. You might
even use IteratorInputFormat like this:
private static class URLInputIterator implements Iterator<Tuple2<Long,
Long>>, Serializable {
private final URL url;
private Iterator<Tuple2<Long, Long>> inner;
private URLInputIterator(URL url) {
this.url = url;
}
private void readObject(ObjectInputStream in) throws IOException,
ClassNotFoundException {
InputStream inputStream = url.openStream();
inner = new BufferedReader(new InputStreamReader(inputStream,
StandardCharsets.UTF_8))
.lines()
.map(line -> {
String[] parts = line.split(";");
return new Tuple2<>(Long.parseLong(parts[0]),
Long.parseLong(parts[1]));
})
.iterator();
}
@Override
public boolean hasNext() {
return inner.hasNext();
}
@Override
public Tuple2<Long, Long> next() {
return inner.next();
}
}
env.fromCollection(new URLInputIterator(new URL("gen_node1", 9999)),
Types.TUPLE(Types.LONG, Types.LONG));
On Fri, Apr 24, 2020 at 9:42 AM Kaan Sancak <ka...@gmail.com> wrote:
> Yes, that sounds like a great idea and actually that's what I am trying to
> do.
>
> Then you configure your analysis job to read from each of these sockets
> with a separate source and union them before feeding them to the actual job?
>
>
> Before trying to open the sockets on the slave nodes, first I have opened
> just one socket at master node, and I also run the generator with one node
> as well. I was able to read the graph, and the run my algorithm without any
> problems. This was a test run to see whatever I can do it.
>
> After, I have opened bunch of sockets on my generators, now I am trying to
> configure Flink to read from those sockets. However, I am having problems
> while trying to assign each task manager to a separate socket. I am
> assuming my problems are related to network binding. In my configuration
> file, jobmanager.rpc.address is set but I have not done
> similar configurations for slave nodes.
>
> Am I on the right track, or is there an easier way to handle this?
>
> I think my point is how to do `read from each of these sockets with a
> separate source` part.
>
> Thanks again
>
> Best
> Kaan
>
>
>
> On Apr 24, 2020, at 3:11 AM, Arvid Heise <ar...@ververica.com> wrote:
>
> Hi Kaan,
>
> sorry, I haven't considered I/O as the bottleneck. I thought a bit more
> about your issue and came to a rather simple solution.
>
> How about you open a socket on each of your generator nodes? Then you
> configure your analysis job to read from each of these sockets with a
> separate source and union them before feeding them to the actual job?
>
> You don't need to modify much on the analysis job and each source can be
> independently read. WDYT?
>
> On Fri, Apr 24, 2020 at 8:46 AM Kaan Sancak <ka...@gmail.com> wrote:
>
>> Thanks for the answer! Also thanks for raising some concerns about my
>> question.
>>
>> Some of the graphs I have been using is larger than 1.5 tb, and I am
>> currently an experiment stage of a project, and I am making modifications
>> to my code and re-runing the experiments again. Currently, on some of the
>> largest graphs I have been using, IO became an issue for me and keeps me
>> wait for couple of hours.
>>
>> Moreover, I have a parallel/distributed graph generator, which I can run
>> on the same set of nodes in my cluster. So what I wanted to do was, to run
>> my Flink program and graph generator at the same time and feed the graph
>> through generator, which should be faster than making IO from the disk. As
>> you said, it is not essential for me to that, but I am trying to see what I
>> am able to do using Flink and how can I solve such problems. I was also
>> using another framework, and faced with the similar problem, I was able to
>> reduce the graph read time from hours to minutes using this method.
>>
>> Do you really have more main memory than disk space?
>>
>>
>> My issue is actually not storage related, I am trying to see how can I
>> reduce the IO time.
>>
>> One trick came to my mind is, creating dummy dataset, and using a map
>> function on the dataset, I can open-up bunch of sockets and listen the
>> generator, and collect the generated data. I am trying to see how it will
>> turn out.
>>
>> Alternatively, if graph generation is rather cheap, you could also try to
>> incorporate it directly into the analysis job.
>>
>>
>> I am not familiar with the analysis jobs. I will look into it.
>>
>> Again, this is actually not a problem, I am just trying to experiment
>> with the framework and see what I can do. I am very new to Flink, so my
>> methods might be wrong. Thanks for the help!
>>
>> Best
>> Kaan
>>
>>
>> On Apr 23, 2020, at 10:51 AM, Arvid Heise <ar...@ververica.com> wrote:
>>
>> Hi Kaan,
>>
>> afaik there is no (easy) way to switch from streaming back to batch API
>> while retaining all data in memory (correct me if I misunderstood).
>>
>> However, from your description, I also have some severe understanding
>> problems. Why can't you dump the data to some file? Do you really have more
>> main memory than disk space? Or do you have no shared memory between your
>> generating cluster and the flink cluster?
>>
>> It almost sounds as if the issue at heart is rather to find a good
>> serialization format on how to store the edges. The 70 billion edges could
>> be stored in an array of id pairs, which amount to ~560 GB uncompressed
>> data if stored in Avro (or any other binary serialization format) when ids
>> are longs. That's not much by today's standards and could also be easily
>> offloaded to S3.
>>
>> Alternatively, if graph generation is rather cheap, you could also try to
>> incorporate it directly into the analysis job.
>>
>> On Wed, Apr 22, 2020 at 2:58 AM Kaan Sancak <ka...@gmail.com> wrote:
>>
>>> Hi,
>>>
>>> I have been running some experiments on large graph data, smallest
>>> graph I have been using is around ~70 billion edges. I have a graph
>>> generator, which generates the graph in parallel and feeds to the running
>>> system. However, it takes a lot of time to read the edges, because even
>>> though the graph generation process is parallel, in Flink I can only listen
>>> from master node (correct me if I am wrong). Another option is dumping the
>>> generated data to a file and reading with readFromCsv, however this is not
>>> feasible in terms of storage management.
>>>
>>> What I want to do is, invoking my graph generator, using ipc/tcp
>>> protocols and reading the generated data from the sockets. Since the graph
>>> data is also generated parallel in each node, I want to make use of ipc,
>>> and read the data in parallel at each node. I made some online digging but
>>> couldn’t find something similar using dataset api. I would be glad if you
>>> have some similar use cases or examples.
>>>
>>> Is it possible to use streaming environment to create the data in
>>> parallel and switch to dataset api?
>>>
>>> Thanks in advance!
>>>
>>> Best
>>> Kaan
>>
>>
>>
>> --
>> Arvid Heise | Senior Java Developer
>> <https://www.ververica.com/>
>>
>> Follow us @VervericaData
>> --
>> Join Flink Forward <https://flink-forward.org/> - The Apache Flink
>> Conference
>> Stream Processing | Event Driven | Real Time
>> --
>> Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
>> --
>> Ververica GmbH
>> Registered at Amtsgericht Charlottenburg: HRB 158244 B
>> Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji
>> (Toni) Cheng
>>
>>
>>
>
> --
> Arvid Heise | Senior Java Developer
> <https://www.ververica.com/>
>
> Follow us @VervericaData
> --
> Join Flink Forward <https://flink-forward.org/> - The Apache Flink
> Conference
> Stream Processing | Event Driven | Real Time
> --
> Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
> --
> Ververica GmbH
> Registered at Amtsgericht Charlottenburg: HRB 158244 B
> Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji
> (Toni) Cheng
>
>
>
--
Arvid Heise | Senior Java Developer
<https://www.ververica.com/>
Follow us @VervericaData
--
Join Flink Forward <https://flink-forward.org/> - The Apache Flink
Conference
Stream Processing | Event Driven | Real Time
--
Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
--
Ververica GmbH
Registered at Amtsgericht Charlottenburg: HRB 158244 B
Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji
(Toni) Cheng
Re: Reading from sockets using dataset api
Posted by Kaan Sancak <ka...@gmail.com>.
Yes, that sounds like a great idea and actually that's what I am trying to do.
> Then you configure your analysis job to read from each of these sockets with a separate source and union them before feeding them to the actual job?
Before trying to open the sockets on the slave nodes, first I have opened just one socket at master node, and I also run the generator with one node as well. I was able to read the graph, and the run my algorithm without any problems. This was a test run to see whatever I can do it.
After, I have opened bunch of sockets on my generators, now I am trying to configure Flink to read from those sockets. However, I am having problems while trying to assign each task manager to a separate socket. I am assuming my problems are related to network binding. In my configuration file, jobmanager.rpc.address is set but I have not done similar configurations for slave nodes.
Am I on the right track, or is there an easier way to handle this?
I think my point is how to do `read from each of these sockets with a separate source` part.
Thanks again
Best
Kaan
> On Apr 24, 2020, at 3:11 AM, Arvid Heise <ar...@ververica.com> wrote:
>
> Hi Kaan,
>
> sorry, I haven't considered I/O as the bottleneck. I thought a bit more about your issue and came to a rather simple solution.
>
> How about you open a socket on each of your generator nodes? Then you configure your analysis job to read from each of these sockets with a separate source and union them before feeding them to the actual job?
>
> You don't need to modify much on the analysis job and each source can be independently read. WDYT?
>
> On Fri, Apr 24, 2020 at 8:46 AM Kaan Sancak <kaansnck@gmail.com <ma...@gmail.com>> wrote:
> Thanks for the answer! Also thanks for raising some concerns about my question.
>
> Some of the graphs I have been using is larger than 1.5 tb, and I am currently an experiment stage of a project, and I am making modifications to my code and re-runing the experiments again. Currently, on some of the largest graphs I have been using, IO became an issue for me and keeps me wait for couple of hours.
>
> Moreover, I have a parallel/distributed graph generator, which I can run on the same set of nodes in my cluster. So what I wanted to do was, to run my Flink program and graph generator at the same time and feed the graph through generator, which should be faster than making IO from the disk. As you said, it is not essential for me to that, but I am trying to see what I am able to do using Flink and how can I solve such problems. I was also using another framework, and faced with the similar problem, I was able to reduce the graph read time from hours to minutes using this method.
>
>> Do you really have more main memory than disk space?
>
> My issue is actually not storage related, I am trying to see how can I reduce the IO time.
>
> One trick came to my mind is, creating dummy dataset, and using a map function on the dataset, I can open-up bunch of sockets and listen the generator, and collect the generated data. I am trying to see how it will turn out.
>
>> Alternatively, if graph generation is rather cheap, you could also try to incorporate it directly into the analysis job.
>
> I am not familiar with the analysis jobs. I will look into it.
>
> Again, this is actually not a problem, I am just trying to experiment with the framework and see what I can do. I am very new to Flink, so my methods might be wrong. Thanks for the help!
>
> Best
> Kaan
>
>
>> On Apr 23, 2020, at 10:51 AM, Arvid Heise <arvid@ververica.com <ma...@ververica.com>> wrote:
>>
>> Hi Kaan,
>>
>> afaik there is no (easy) way to switch from streaming back to batch API while retaining all data in memory (correct me if I misunderstood).
>>
>> However, from your description, I also have some severe understanding problems. Why can't you dump the data to some file? Do you really have more main memory than disk space? Or do you have no shared memory between your generating cluster and the flink cluster?
>>
>> It almost sounds as if the issue at heart is rather to find a good serialization format on how to store the edges. The 70 billion edges could be stored in an array of id pairs, which amount to ~560 GB uncompressed data if stored in Avro (or any other binary serialization format) when ids are longs. That's not much by today's standards and could also be easily offloaded to S3.
>>
>> Alternatively, if graph generation is rather cheap, you could also try to incorporate it directly into the analysis job.
>>
>> On Wed, Apr 22, 2020 at 2:58 AM Kaan Sancak <kaansnck@gmail.com <ma...@gmail.com>> wrote:
>> Hi,
>>
>> I have been running some experiments on large graph data, smallest graph I have been using is around ~70 billion edges. I have a graph generator, which generates the graph in parallel and feeds to the running system. However, it takes a lot of time to read the edges, because even though the graph generation process is parallel, in Flink I can only listen from master node (correct me if I am wrong). Another option is dumping the generated data to a file and reading with readFromCsv, however this is not feasible in terms of storage management.
>>
>> What I want to do is, invoking my graph generator, using ipc/tcp protocols and reading the generated data from the sockets. Since the graph data is also generated parallel in each node, I want to make use of ipc, and read the data in parallel at each node. I made some online digging but couldn’t find something similar using dataset api. I would be glad if you have some similar use cases or examples.
>>
>> Is it possible to use streaming environment to create the data in parallel and switch to dataset api?
>>
>> Thanks in advance!
>>
>> Best
>> Kaan
>>
>>
>> --
>> Arvid Heise | Senior Java Developer
>> <https://www.ververica.com/>
>> Follow us @VervericaData
>> --
>> Join Flink Forward <https://flink-forward.org/> - The Apache Flink Conference
>> Stream Processing | Event Driven | Real Time
>> --
>> Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
>> --
>> Ververica GmbH
>> Registered at Amtsgericht Charlottenburg: HRB 158244 B
>> Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji (Toni) Cheng
>
>
>
> --
> Arvid Heise | Senior Java Developer
> <https://www.ververica.com/>
> Follow us @VervericaData
> --
> Join Flink Forward <https://flink-forward.org/> - The Apache Flink Conference
> Stream Processing | Event Driven | Real Time
> --
> Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
> --
> Ververica GmbH
> Registered at Amtsgericht Charlottenburg: HRB 158244 B
> Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji (Toni) Cheng
Re: Reading from sockets using dataset api
Posted by Arvid Heise <ar...@ververica.com>.
Hi Kaan,
sorry, I haven't considered I/O as the bottleneck. I thought a bit more
about your issue and came to a rather simple solution.
How about you open a socket on each of your generator nodes? Then you
configure your analysis job to read from each of these sockets with a
separate source and union them before feeding them to the actual job?
You don't need to modify much on the analysis job and each source can be
independently read. WDYT?
On Fri, Apr 24, 2020 at 8:46 AM Kaan Sancak <ka...@gmail.com> wrote:
> Thanks for the answer! Also thanks for raising some concerns about my
> question.
>
> Some of the graphs I have been using is larger than 1.5 tb, and I am
> currently an experiment stage of a project, and I am making modifications
> to my code and re-runing the experiments again. Currently, on some of the
> largest graphs I have been using, IO became an issue for me and keeps me
> wait for couple of hours.
>
> Moreover, I have a parallel/distributed graph generator, which I can run
> on the same set of nodes in my cluster. So what I wanted to do was, to run
> my Flink program and graph generator at the same time and feed the graph
> through generator, which should be faster than making IO from the disk. As
> you said, it is not essential for me to that, but I am trying to see what I
> am able to do using Flink and how can I solve such problems. I was also
> using another framework, and faced with the similar problem, I was able to
> reduce the graph read time from hours to minutes using this method.
>
> Do you really have more main memory than disk space?
>
>
> My issue is actually not storage related, I am trying to see how can I
> reduce the IO time.
>
> One trick came to my mind is, creating dummy dataset, and using a map
> function on the dataset, I can open-up bunch of sockets and listen the
> generator, and collect the generated data. I am trying to see how it will
> turn out.
>
> Alternatively, if graph generation is rather cheap, you could also try to
> incorporate it directly into the analysis job.
>
>
> I am not familiar with the analysis jobs. I will look into it.
>
> Again, this is actually not a problem, I am just trying to experiment with
> the framework and see what I can do. I am very new to Flink, so my methods
> might be wrong. Thanks for the help!
>
> Best
> Kaan
>
>
> On Apr 23, 2020, at 10:51 AM, Arvid Heise <ar...@ververica.com> wrote:
>
> Hi Kaan,
>
> afaik there is no (easy) way to switch from streaming back to batch API
> while retaining all data in memory (correct me if I misunderstood).
>
> However, from your description, I also have some severe understanding
> problems. Why can't you dump the data to some file? Do you really have more
> main memory than disk space? Or do you have no shared memory between your
> generating cluster and the flink cluster?
>
> It almost sounds as if the issue at heart is rather to find a good
> serialization format on how to store the edges. The 70 billion edges could
> be stored in an array of id pairs, which amount to ~560 GB uncompressed
> data if stored in Avro (or any other binary serialization format) when ids
> are longs. That's not much by today's standards and could also be easily
> offloaded to S3.
>
> Alternatively, if graph generation is rather cheap, you could also try to
> incorporate it directly into the analysis job.
>
> On Wed, Apr 22, 2020 at 2:58 AM Kaan Sancak <ka...@gmail.com> wrote:
>
>> Hi,
>>
>> I have been running some experiments on large graph data, smallest graph
>> I have been using is around ~70 billion edges. I have a graph generator,
>> which generates the graph in parallel and feeds to the running system.
>> However, it takes a lot of time to read the edges, because even though the
>> graph generation process is parallel, in Flink I can only listen from
>> master node (correct me if I am wrong). Another option is dumping the
>> generated data to a file and reading with readFromCsv, however this is not
>> feasible in terms of storage management.
>>
>> What I want to do is, invoking my graph generator, using ipc/tcp
>> protocols and reading the generated data from the sockets. Since the graph
>> data is also generated parallel in each node, I want to make use of ipc,
>> and read the data in parallel at each node. I made some online digging but
>> couldn’t find something similar using dataset api. I would be glad if you
>> have some similar use cases or examples.
>>
>> Is it possible to use streaming environment to create the data in
>> parallel and switch to dataset api?
>>
>> Thanks in advance!
>>
>> Best
>> Kaan
>
>
>
> --
> Arvid Heise | Senior Java Developer
> <https://www.ververica.com/>
>
> Follow us @VervericaData
> --
> Join Flink Forward <https://flink-forward.org/> - The Apache Flink
> Conference
> Stream Processing | Event Driven | Real Time
> --
> Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
> --
> Ververica GmbH
> Registered at Amtsgericht Charlottenburg: HRB 158244 B
> Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji
> (Toni) Cheng
>
>
>
--
Arvid Heise | Senior Java Developer
<https://www.ververica.com/>
Follow us @VervericaData
--
Join Flink Forward <https://flink-forward.org/> - The Apache Flink
Conference
Stream Processing | Event Driven | Real Time
--
Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
--
Ververica GmbH
Registered at Amtsgericht Charlottenburg: HRB 158244 B
Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji
(Toni) Cheng
Re: Reading from sockets using dataset api
Posted by Kaan Sancak <ka...@gmail.com>.
Thanks for the answer! Also thanks for raising some concerns about my question.
Some of the graphs I have been using is larger than 1.5 tb, and I am currently an experiment stage of a project, and I am making modifications to my code and re-runing the experiments again. Currently, on some of the largest graphs I have been using, IO became an issue for me and keeps me wait for couple of hours.
Moreover, I have a parallel/distributed graph generator, which I can run on the same set of nodes in my cluster. So what I wanted to do was, to run my Flink program and graph generator at the same time and feed the graph through generator, which should be faster than making IO from the disk. As you said, it is not essential for me to that, but I am trying to see what I am able to do using Flink and how can I solve such problems. I was also using another framework, and faced with the similar problem, I was able to reduce the graph read time from hours to minutes using this method.
> Do you really have more main memory than disk space?
My issue is actually not storage related, I am trying to see how can I reduce the IO time.
One trick came to my mind is, creating dummy dataset, and using a map function on the dataset, I can open-up bunch of sockets and listen the generator, and collect the generated data. I am trying to see how it will turn out.
> Alternatively, if graph generation is rather cheap, you could also try to incorporate it directly into the analysis job.
I am not familiar with the analysis jobs. I will look into it.
Again, this is actually not a problem, I am just trying to experiment with the framework and see what I can do. I am very new to Flink, so my methods might be wrong. Thanks for the help!
Best
Kaan
> On Apr 23, 2020, at 10:51 AM, Arvid Heise <ar...@ververica.com> wrote:
>
> Hi Kaan,
>
> afaik there is no (easy) way to switch from streaming back to batch API while retaining all data in memory (correct me if I misunderstood).
>
> However, from your description, I also have some severe understanding problems. Why can't you dump the data to some file? Do you really have more main memory than disk space? Or do you have no shared memory between your generating cluster and the flink cluster?
>
> It almost sounds as if the issue at heart is rather to find a good serialization format on how to store the edges. The 70 billion edges could be stored in an array of id pairs, which amount to ~560 GB uncompressed data if stored in Avro (or any other binary serialization format) when ids are longs. That's not much by today's standards and could also be easily offloaded to S3.
>
> Alternatively, if graph generation is rather cheap, you could also try to incorporate it directly into the analysis job.
>
> On Wed, Apr 22, 2020 at 2:58 AM Kaan Sancak <kaansnck@gmail.com <ma...@gmail.com>> wrote:
> Hi,
>
> I have been running some experiments on large graph data, smallest graph I have been using is around ~70 billion edges. I have a graph generator, which generates the graph in parallel and feeds to the running system. However, it takes a lot of time to read the edges, because even though the graph generation process is parallel, in Flink I can only listen from master node (correct me if I am wrong). Another option is dumping the generated data to a file and reading with readFromCsv, however this is not feasible in terms of storage management.
>
> What I want to do is, invoking my graph generator, using ipc/tcp protocols and reading the generated data from the sockets. Since the graph data is also generated parallel in each node, I want to make use of ipc, and read the data in parallel at each node. I made some online digging but couldn’t find something similar using dataset api. I would be glad if you have some similar use cases or examples.
>
> Is it possible to use streaming environment to create the data in parallel and switch to dataset api?
>
> Thanks in advance!
>
> Best
> Kaan
>
>
> --
> Arvid Heise | Senior Java Developer
> <https://www.ververica.com/>
> Follow us @VervericaData
> --
> Join Flink Forward <https://flink-forward.org/> - The Apache Flink Conference
> Stream Processing | Event Driven | Real Time
> --
> Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
> --
> Ververica GmbH
> Registered at Amtsgericht Charlottenburg: HRB 158244 B
> Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji (Toni) Cheng
Re: Reading from sockets using dataset api
Posted by Arvid Heise <ar...@ververica.com>.
Hi Kaan,
afaik there is no (easy) way to switch from streaming back to batch API
while retaining all data in memory (correct me if I misunderstood).
However, from your description, I also have some severe understanding
problems. Why can't you dump the data to some file? Do you really have more
main memory than disk space? Or do you have no shared memory between your
generating cluster and the flink cluster?
It almost sounds as if the issue at heart is rather to find a good
serialization format on how to store the edges. The 70 billion edges could
be stored in an array of id pairs, which amount to ~560 GB uncompressed
data if stored in Avro (or any other binary serialization format) when ids
are longs. That's not much by today's standards and could also be easily
offloaded to S3.
Alternatively, if graph generation is rather cheap, you could also try to
incorporate it directly into the analysis job.
On Wed, Apr 22, 2020 at 2:58 AM Kaan Sancak <ka...@gmail.com> wrote:
> Hi,
>
> I have been running some experiments on large graph data, smallest graph
> I have been using is around ~70 billion edges. I have a graph generator,
> which generates the graph in parallel and feeds to the running system.
> However, it takes a lot of time to read the edges, because even though the
> graph generation process is parallel, in Flink I can only listen from
> master node (correct me if I am wrong). Another option is dumping the
> generated data to a file and reading with readFromCsv, however this is not
> feasible in terms of storage management.
>
> What I want to do is, invoking my graph generator, using ipc/tcp
> protocols and reading the generated data from the sockets. Since the graph
> data is also generated parallel in each node, I want to make use of ipc,
> and read the data in parallel at each node. I made some online digging but
> couldn’t find something similar using dataset api. I would be glad if you
> have some similar use cases or examples.
>
> Is it possible to use streaming environment to create the data in parallel
> and switch to dataset api?
>
> Thanks in advance!
>
> Best
> Kaan
--
Arvid Heise | Senior Java Developer
<https://www.ververica.com/>
Follow us @VervericaData
--
Join Flink Forward <https://flink-forward.org/> - The Apache Flink
Conference
Stream Processing | Event Driven | Real Time
--
Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
--
Ververica GmbH
Registered at Amtsgericht Charlottenburg: HRB 158244 B
Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji
(Toni) Cheng