flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Flavio Pompermaier <pomperma...@okkam.it>
Subject Re: Read once input data?
Date Wed, 17 Feb 2016 09:44:58 GMT
In my use case I though to persist the dataset to reuse on Tachyon in order
to speed up its reading..do you think it could help?

On Tue, Feb 16, 2016 at 10:28 PM, Saliya Ekanayake <esaliya@gmail.com>
wrote:

> Thank you. I'll check this
>
> On Tue, Feb 16, 2016 at 4:01 PM, Fabian Hueske <fhueske@gmail.com> wrote:
>
>> Broadcasted DataSets are stored on the JVM heap of each task manager (but
>> shared among multiple slots on the same TM), hence the size restriction.
>>
>> There are two ways to retrieve a DataSet (such as the result of a reduce).
>> 1) if you want to fetch the result into your client program use
>> DataSet.collect(). This immediately triggers an execution and fetches the
>> result from the cluster.
>> 2) if you want to use the result for a computation in the cluster use
>> broadcast sets as described above.
>>
>> 2016-02-16 21:54 GMT+01:00 Saliya Ekanayake <esaliya@gmail.com>:
>>
>>> Thank you, yes, this makes sense. The broadcasted data in my case would
>>> a large array of 3D coordinates,
>>>
>>> On a side note, how can I take the output from a reduce function? I can
>>> see methods to write it to a given output, but is it possible to retrieve
>>> the reduced result back to the program - like a double value representing
>>> the average in the previous example.
>>>
>>>
>>> On Tue, Feb 16, 2016 at 3:47 PM, Fabian Hueske <fhueske@gmail.com>
>>> wrote:
>>>
>>>> You can use so-called BroadcastSets to send any sufficiently small
>>>> DataSet (such as a computed average) to any other function and use it there.
>>>> However, in your case you'll end up with a data flow that branches (at
>>>> the source) and merges again (when the average is send to the second map).
>>>> Such patterns can cause deadlocks and can therefore not be pipelined
>>>> which means that the data before the branch is written to disk and read
>>>> again.
>>>> In your case it might be even better to read the data twice instead of
>>>> reading, writing, and reading it.
>>>>
>>>> Fabian
>>>>
>>>> 2016-02-16 21:15 GMT+01:00 Saliya Ekanayake <esaliya@gmail.com>:
>>>>
>>>>> I looked at the samples and I think what you meant is clear, but I
>>>>> didn't find a solution for my need. In my case, I want to use the result
>>>>> from first map operation before I can apply the second map on the
>>>>> *same* data set. For simplicity, let's say I've a bunch of short
>>>>> values represented as my data set. Then I need to find their average,
so I
>>>>> use a map and reduce. Then I want to map these short values with another
>>>>> function, but it needs that average computed in the beginning to work
>>>>> correctly.
>>>>>
>>>>> Is this possible without doing multiple reads of the input data to
>>>>> create the same dataset?
>>>>>
>>>>> Thank you,
>>>>> saliya
>>>>>
>>>>> On Tue, Feb 16, 2016 at 12:03 PM, Fabian Hueske <fhueske@gmail.com>
>>>>> wrote:
>>>>>
>>>>>> Yes, if you implement both maps in a single job, data is read once.
>>>>>>
>>>>>> 2016-02-16 15:53 GMT+01:00 Saliya Ekanayake <esaliya@gmail.com>:
>>>>>>
>>>>>>> Fabian,
>>>>>>>
>>>>>>> I've a quick follow-up question on what you suggested. When
>>>>>>> streaming the same data through different maps, were you implying
that
>>>>>>> everything goes as single job in Flink, so data read happens
only once?
>>>>>>>
>>>>>>> Thanks,
>>>>>>> Saliya
>>>>>>>
>>>>>>> On Mon, Feb 15, 2016 at 3:58 PM, Fabian Hueske <fhueske@gmail.com>
>>>>>>> wrote:
>>>>>>>
>>>>>>>> It is not possible to "pin" data sets in memory, yet.
>>>>>>>> However, you can stream the same data set through two different
>>>>>>>> mappers at the same time.
>>>>>>>>
>>>>>>>> For instance you can have a job like:
>>>>>>>>
>>>>>>>>                  /---> Map 1 --> SInk1
>>>>>>>> Source --<
>>>>>>>>                  \---> Map 2 --> SInk2
>>>>>>>>
>>>>>>>> and execute it at once.
>>>>>>>> For that you define you data flow and call execute once after
all
>>>>>>>> sinks have been created.
>>>>>>>>
>>>>>>>> Best, Fabian
>>>>>>>>
>>>>>>>> 2016-02-15 21:32 GMT+01:00 Saliya Ekanayake <esaliya@gmail.com>:
>>>>>>>>
>>>>>>>>> Fabian,
>>>>>>>>>
>>>>>>>>> count() was just an example. What I would like to do
is say run
>>>>>>>>> two map operations on the dataset (ds). Each map will
have it's own
>>>>>>>>> reduction, so is there a way to avoid creating two jobs
for such scenario?
>>>>>>>>>
>>>>>>>>> The reason is, reading these binary matrices are expensive.
In our
>>>>>>>>> current MPI implementation, I am using memory maps for
faster loading and
>>>>>>>>> reuse.
>>>>>>>>>
>>>>>>>>> Thank you,
>>>>>>>>> Saliya
>>>>>>>>>
>>>>>>>>> On Mon, Feb 15, 2016 at 3:15 PM, Fabian Hueske <fhueske@gmail.com>
>>>>>>>>> wrote:
>>>>>>>>>
>>>>>>>>>> Hi,
>>>>>>>>>>
>>>>>>>>>> it looks like you are executing two distinct Flink
jobs.
>>>>>>>>>> DataSet.count() triggers the execution of a new job.
If you have
>>>>>>>>>> an execute() call in your program, this will lead
to two Flink jobs being
>>>>>>>>>> executed.
>>>>>>>>>> It is not possible to share state among these jobs.
>>>>>>>>>>
>>>>>>>>>> Maybe you should add a custom count implementation
(using a
>>>>>>>>>> ReduceFunction) which is executed in the same program
as the other
>>>>>>>>>> ReduceFunction.
>>>>>>>>>>
>>>>>>>>>> Best, Fabian
>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>> 2016-02-15 21:05 GMT+01:00 Saliya Ekanayake <esaliya@gmail.com>:
>>>>>>>>>>
>>>>>>>>>>> Hi,
>>>>>>>>>>>
>>>>>>>>>>> I see that an InputFormat's open() and nextRecord()
methods get
>>>>>>>>>>> called for each terminal operation on a given
dataset using that particular
>>>>>>>>>>> InputFormat. Is it possible to avoid this - possibly
using some caching
>>>>>>>>>>> technique in Flink?
>>>>>>>>>>>
>>>>>>>>>>> For example, I've some code like below and I
see for both the
>>>>>>>>>>> last two statements (reduce() and count()) the
above methods in the input
>>>>>>>>>>> format get called. Btw. this is a custom input
format I wrote to represent
>>>>>>>>>>> a binary matrix stored as Short values.
>>>>>>>>>>>
>>>>>>>>>>> ShortMatrixInputFormat smif = new ShortMatrixInputFormat();
>>>>>>>>>>>
>>>>>>>>>>> DataSet<Short[]> ds = env.createInput(smif,
BasicArrayTypeInfo.SHORT_ARRAY_TYPE_INFO);
>>>>>>>>>>>
>>>>>>>>>>> MapOperator<Short[], DoubleStatistics>
op = ds.map(...)
>>>>>>>>>>>
>>>>>>>>>>> *op.reduce(...)*
>>>>>>>>>>>
>>>>>>>>>>> *op.count(...)*
>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>> Thank you,
>>>>>>>>>>> Saliya
>>>>>>>>>>> --
>>>>>>>>>>> Saliya Ekanayake
>>>>>>>>>>> Ph.D. Candidate | Research Assistant
>>>>>>>>>>> School of Informatics and Computing | Digital
Science Center
>>>>>>>>>>> Indiana University, Bloomington
>>>>>>>>>>> Cell 812-391-4914
>>>>>>>>>>> http://saliya.org
>>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> --
>>>>>>>>> Saliya Ekanayake
>>>>>>>>> Ph.D. Candidate | Research Assistant
>>>>>>>>> School of Informatics and Computing | Digital Science
Center
>>>>>>>>> Indiana University, Bloomington
>>>>>>>>> Cell 812-391-4914
>>>>>>>>> http://saliya.org
>>>>>>>>>
>>>>>>>>
>>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>> --
>>>>>>> Saliya Ekanayake
>>>>>>> Ph.D. Candidate | Research Assistant
>>>>>>> School of Informatics and Computing | Digital Science Center
>>>>>>> Indiana University, Bloomington
>>>>>>> Cell 812-391-4914
>>>>>>> http://saliya.org
>>>>>>>
>>>>>>
>>>>>>
>>>>>
>>>>>
>>>>> --
>>>>> Saliya Ekanayake
>>>>> Ph.D. Candidate | Research Assistant
>>>>> School of Informatics and Computing | Digital Science Center
>>>>> Indiana University, Bloomington
>>>>> Cell 812-391-4914
>>>>> http://saliya.org
>>>>>
>>>>
>>>>
>>>
>>>
>>> --
>>> Saliya Ekanayake
>>> Ph.D. Candidate | Research Assistant
>>> School of Informatics and Computing | Digital Science Center
>>> Indiana University, Bloomington
>>> Cell 812-391-4914
>>> http://saliya.org
>>>
>>
>>
>
>
> --
> Saliya Ekanayake
> Ph.D. Candidate | Research Assistant
> School of Informatics and Computing | Digital Science Center
> Indiana University, Bloomington
> Cell 812-391-4914
> http://saliya.org
>

Mime
View raw message