You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@mrql.apache.org by Eldon Carman <ec...@ucr.edu> on 2014/10/22 23:42:27 UTC
MRQL and Flink configuration
Hi Guys,
I am working on setting up Flink and MRQL. I have been able to get the
system up and running. My question is about configuring them to utilize all
the available resources. My tests are showing only 50% of one node is being
utilized (basically 2 virtual cpus).
Configuration:
master - NameNode, Client, CliFrontend, JobHistoryServer, ResourceManager
slave(1,2) - YarnTaskManagerRunner(x4),DataNode,NodeManager
Flink:
4 TaskManagers each with access to 4 virtual cpus and 1 GB of memory
Actual Node Resources:
4 cores
8 GB memory
Only two virtual cpus are being used to process the MRQL query. I see you
have partial documentation up for Flink. (Need a working example.) Any
suggestions on how to use more of the Flink TaskManagers?
Thanks
Eldon
Re: MRQL and Flink configuration
Posted by Leonidas Fegaras <fe...@cse.uta.edu>.
Hi Eldon,
I think this particular query has been implemented better in the MRQL
MapReduce evaluation engine than the one for Flink.
This query is translated to a MapAggregateReduce2 operation, which is a
join combined with a total aggregation. In MapReduce, it is implemented
as a reduce-side join that does the total aggregation at each reducer.
So each reducer produces only one aggregation result. These results are
then collected and aggregated. In Flink mode, this operation is
implemented as a coGroup followed by total aggregation, which is
inefficient because coGroup produces a very big result. I will change
the MRQL Flink evaluator to do aggregations during coGroup.
Flink (as well as Spark) is typically faster than MapReduce when you
have multiple MapReduce jobs or iterations.
Leonidas
On 10/24/2014 12:20 AM, Eldon Carman wrote:
> Thanks for your support.
>
> I used your examples to use all the available cores in a series of
> experiments. The experiment was using the MRQL from git, Flink 0.6.1,
> and Hadoop 2.5.1 on a two node cluster each with four cores (different
> values from the previous example). I observed that Flink JobManager
> showed all eight containers being used. The query time was about two
> times slower than the same experiment on Hadoop 1.2.1 using MapReduce.
> I would have expected Flink to be faster. During the Flink test, the
> CPUs were near 100% and the memory was about 50% per thread. Just
> wanted to get your feedback on these results?
>
> Thanks
> Eldon
>
>
> Test Scenario: 2 nodes with 8GB of memory
> Dataset Size: 7.2GB per node
>
> Flink Configuration: -n 8 -tm 1024
> MRQL Configuration: -nodes 8
>
> MapReduce Query Time: 460s
> Flink Query Time: 1200s
>
> MRQL query:
> min(
> select (toInt(text(sensors.value)))
> from sensors in source(xml, args[0], {"data"}),
> stations in source(xml, args[1], {"station"}),
> l in stations.locationLabels
> where text(stations.id <http://stations.id/>) = text(sensors.station)
> and toInt(substring(text(sensors.date), 0, 4)) = 2001
> and text(sensors.dataType) = "TMIN"
> and text(l.id <http://l.id/>) = "FIPS:US"
> ) / 10;
>
> If you would like, I can e-mail you part of the dataset.
>
> On Wed, Oct 22, 2014 at 3:59 PM, Leonidas Fegaras <fegaras@cse.uta.edu
> <ma...@cse.uta.edu>> wrote:
>
> Hi Eldon,
> I am assuming you are using flink 0.6.1 on a yarn cluster.
> You can see the Flink startup parameters by running:
> ${FLINK_HOME}/bin/yarn-session.sh
> Basically, you can allocate one cpu for the JobManager and the
> rest (4*4-1=15 cpus) for the TaskManagers.
> Also, you may need to set the TaskManager memory using -tm.
> That is, you can start Flink using:
> ${FLINK_HOME}/bin/yarn-session.sh -n 15 -tm 2000
> You can verify this by looking at: http://...:8088/cluster/nodes
> (it should use all containers and all memory)
> and by looking at the Flink JobManager Web Interface (you can get
> the link from: http://...:8088/cluster/apps ).
> Then, you run MRQL on Flink using:
> bin/mrql.flink -dist -nodes 15 ...
> (use -info for log info).
> Note that for Flink 0.7.0, you can specify the number of slots
> using -s.
> Leonidas
>
>
>
> On 10/22/2014 04:42 PM, Eldon Carman wrote:
>
> Hi Guys,
>
> I am working on setting up Flink and MRQL. I have been able to
> get the system up and running. My question is about
> configuring them to utilize all the available resources. My
> tests are showing only 50% of one node is being utilized
> (basically 2 virtual cpus).
>
> Configuration:
> master - NameNode, Client, CliFrontend, JobHistoryServer,
> ResourceManager
> slave(1,2) - YarnTaskManagerRunner(x4),DataNode,NodeManager
>
> Flink:
> 4 TaskManagers each with access to 4 virtual cpus and 1 GB of
> memory
>
> Actual Node Resources:
> 4 cores
> 8 GB memory
>
> Only two virtual cpus are being used to process the MRQL
> query. I see you have partial documentation up for Flink.
> (Need a working example.) Any suggestions on how to use more
> of the Flink TaskManagers?
>
> Thanks
> Eldon
>
>
>
Re: MRQL and Flink configuration
Posted by Eldon Carman <ec...@ucr.edu>.
Thanks for your support.
I used your examples to use all the available cores in a series of
experiments. The experiment was using the MRQL from git, Flink 0.6.1, and
Hadoop 2.5.1 on a two node cluster each with four cores (different values
from the previous example). I observed that Flink JobManager showed all
eight containers being used. The query time was about two times slower than
the same experiment on Hadoop 1.2.1 using MapReduce. I would have expected
Flink to be faster. During the Flink test, the CPUs were near 100% and the
memory was about 50% per thread. Just wanted to get your feedback on these
results?
Thanks
Eldon
Test Scenario: 2 nodes with 8GB of memory
Dataset Size: 7.2GB per node
Flink Configuration: -n 8 -tm 1024
MRQL Configuration: -nodes 8
MapReduce Query Time: 460s
Flink Query Time: 1200s
MRQL query:
min(
select (toInt(text(sensors.value)))
from sensors in source(xml, args[0], {"data"}),
stations in source(xml, args[1], {"station"}),
l in stations.locationLabels
where text(stations.id) = text(sensors.station)
and toInt(substring(text(sensors.date), 0, 4)) = 2001
and text(sensors.dataType) = "TMIN"
and text(l.id) = "FIPS:US"
) / 10;
If you would like, I can e-mail you part of the dataset.
On Wed, Oct 22, 2014 at 3:59 PM, Leonidas Fegaras <fe...@cse.uta.edu>
wrote:
> Hi Eldon,
> I am assuming you are using flink 0.6.1 on a yarn cluster.
> You can see the Flink startup parameters by running:
> ${FLINK_HOME}/bin/yarn-session.sh
> Basically, you can allocate one cpu for the JobManager and the rest
> (4*4-1=15 cpus) for the TaskManagers.
> Also, you may need to set the TaskManager memory using -tm.
> That is, you can start Flink using:
> ${FLINK_HOME}/bin/yarn-session.sh -n 15 -tm 2000
> You can verify this by looking at: http://...:8088/cluster/nodes (it
> should use all containers and all memory)
> and by looking at the Flink JobManager Web Interface (you can get the link
> from: http://...:8088/cluster/apps ).
> Then, you run MRQL on Flink using:
> bin/mrql.flink -dist -nodes 15 ...
> (use -info for log info).
> Note that for Flink 0.7.0, you can specify the number of slots using -s.
> Leonidas
>
>
>
> On 10/22/2014 04:42 PM, Eldon Carman wrote:
>
>> Hi Guys,
>>
>> I am working on setting up Flink and MRQL. I have been able to get the
>> system up and running. My question is about configuring them to utilize all
>> the available resources. My tests are showing only 50% of one node is being
>> utilized (basically 2 virtual cpus).
>>
>> Configuration:
>> master - NameNode, Client, CliFrontend, JobHistoryServer, ResourceManager
>> slave(1,2) - YarnTaskManagerRunner(x4),DataNode,NodeManager
>>
>> Flink:
>> 4 TaskManagers each with access to 4 virtual cpus and 1 GB of memory
>>
>> Actual Node Resources:
>> 4 cores
>> 8 GB memory
>>
>> Only two virtual cpus are being used to process the MRQL query. I see you
>> have partial documentation up for Flink. (Need a working example.) Any
>> suggestions on how to use more of the Flink TaskManagers?
>>
>> Thanks
>> Eldon
>>
>
>
Re: MRQL and Flink configuration
Posted by Leonidas Fegaras <fe...@cse.uta.edu>.
Hi Eldon,
I am assuming you are using flink 0.6.1 on a yarn cluster.
You can see the Flink startup parameters by running:
${FLINK_HOME}/bin/yarn-session.sh
Basically, you can allocate one cpu for the JobManager and the rest
(4*4-1=15 cpus) for the TaskManagers.
Also, you may need to set the TaskManager memory using -tm.
That is, you can start Flink using:
${FLINK_HOME}/bin/yarn-session.sh -n 15 -tm 2000
You can verify this by looking at: http://...:8088/cluster/nodes (it
should use all containers and all memory)
and by looking at the Flink JobManager Web Interface (you can get the
link from: http://...:8088/cluster/apps ).
Then, you run MRQL on Flink using:
bin/mrql.flink -dist -nodes 15 ...
(use -info for log info).
Note that for Flink 0.7.0, you can specify the number of slots using -s.
Leonidas
On 10/22/2014 04:42 PM, Eldon Carman wrote:
> Hi Guys,
>
> I am working on setting up Flink and MRQL. I have been able to get the
> system up and running. My question is about configuring them to
> utilize all the available resources. My tests are showing only 50% of
> one node is being utilized (basically 2 virtual cpus).
>
> Configuration:
> master - NameNode, Client, CliFrontend, JobHistoryServer, ResourceManager
> slave(1,2) - YarnTaskManagerRunner(x4),DataNode,NodeManager
>
> Flink:
> 4 TaskManagers each with access to 4 virtual cpus and 1 GB of memory
>
> Actual Node Resources:
> 4 cores
> 8 GB memory
>
> Only two virtual cpus are being used to process the MRQL query. I see
> you have partial documentation up for Flink. (Need a working example.)
> Any suggestions on how to use more of the Flink TaskManagers?
>
> Thanks
> Eldon