flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Ufuk Celebi <...@apache.org>
Subject Re: Too few memory segments provided. Hash Table needs at least 33 memory segments.
Date Mon, 14 Nov 2016 09:26:18 GMT
What do the TaskManager logs say wrt to allocation of managed memory?

Something like:

Limiting managed memory to ... of the currently free heap space ..., memory will be allocated
lazily.

What else did you configure in flink-conf?

Looping in Greg and Vasia who maintain Gelly and are most-familiar with the internals.

– Ufuk


On 8 November 2016 at 22:35:22, Miguel Coimbra (miguel.e.coimbra@gmail.com) wrote:
> Dear community,
>  
> I have a problem which I hope you'll be able to help with.
> I apologize in advance for the verbosity of the post.
> I am running the Flink standalone cluster (not even storing to the
> filesystem) with 2 Docker containers.
>  
> I set the image of the Dockerfile for Flink 1.1.2, which was the same
> version of the main class in the .jar
> The Docker image was configured to use Java 8, which is what the project's
> pom.xml requires as well.
> I have also edited the TaskManager conf/flink-con.yaml to have the
> following values:
>  
> ....
> taskmanager.heap.mb: 7512
> ....
> taskmanager.network.numberOfBuffers: 16048
> ....
>  
>  
> Properties of this host/docker setup:
> - host machine has *256 GB *of RAM
> - job manager container is running with default flink config
> - task manager has *7.5 GB *of memory available
> - task manager number of buffers is *16048 *which is very generous compared
> to the default value
>  
> I am testing on the SNAP DBLP dataset:
> https://snap.stanford.edu/data/com-DBLP.html
> It has:
>  
> 317080 nodes
> 1049866 edges
>  
> These are the relevant parts of the pom.xml of the project:
> *(note: the project executes without error for local executions without the
> cluster)*
>  
> ....
>  
> UTF-8
>  
> UTF-8  
> 1.8
> 1.8
> 1.1.2
>  
> .....
>  
>  
> org.apache.flink
> flink-java
> ${flink.version}
>  
>  
> org.apache.flink
> flink-core
> ${flink.version}
>  
>  
> org.apache.flink
> flink-streaming-java_2.10
> ${flink.version}
>  
>  
> org.apache.flink
> flink-clients_2.10
> ${flink.version}
>  
>  
> org.apache.flink
> flink-gelly_2.10
> ${flink.version}
>  
>  
> junit
> junit
> 3.8.1
> test
>  
>  
>  
> I am running (what I believe to be) a simple Gelly application, performing
> the ConnectedComponents algorithm with 30 iterations:
>  
> public static void main(String[] args) {
> final ExecutionEnvironment env =
> ExecutionEnvironment.getExecutionEnvironment();
>  
>  
> final String dataPath = args[0];
>  
> final DataSet> edgeTuples =
> env.readCsvFile(dataPath)
> .fieldDelimiter("\t") // node IDs are separated by spaces
> .ignoreComments("#") // comments start with "%"
> .types(Long.class, Long.class);
>  
> try {
> System.out.println("Tuple size: " + edgeTuples.count());
> } catch (Exception e1) {
> e1.printStackTrace();
> }
>  
> /*
> * @param the key type for edge and vertex identifiers
> * @param the value type for vertices
> * @param the value type for edges
> * public class Graph
> */
>  
>  
> final Graph graph = Graph.fromTuple2DataSet(
> edgeTuples,
> new MapFunction() {
> private static final long serialVersionUID =
> 8713516577419451509L;
> public Long map(Long value) {
> return value;
> }
> },
> env
> );
>  
>  
> try {
> /**
> * @param key type
> * @param vertex value type
> * @param edge value type
> * @param the return type
>  
> class ConnectedComponents, EV>
> implements GraphAlgorithm>>
> */
>  
> DataSet> verticesWithComponents =
> graph.run(new ConnectedComponents(30));
> System.out.println("Component count: " +
> verticesWithComponents.count());
> } catch (Exception e) {
> e.printStackTrace();
> }
> }
>  
>  
> However, the following is output on the host machine on execution:
>  
> docker exec -it $(docker ps --filter name=jobmanager --format={{.ID}})
> flink run -m 3de7625b8e28:6123 -c flink.graph.example.App
> /home/myuser/flink-graph-example-0.0.1-SNAPSHOT.jar
> /home/myuser/com-dblp.ungraph.txt
>  
> Cluster configuration: Standalone cluster with JobManager at /
> 172.19.0.2:6123
> Using address 172.19.0.2:6123 to connect to JobManager.
> JobManager web interface address http://172.19.0.2:8081
> Starting execution of program
> Submitting job with JobID: fd6a12896b749e9ed439bbb196c6aaae. Waiting for
> job completion.
> Connected to JobManager at Actor[akka.tcp://
> flink@172.19.0.2:6123/user/jobmanager#-658812967]
>  
> 11/08/2016 21:22:44 DataSource (at main(App.java:25)
> (org.apache.flink.api.java.io.TupleCsvInputFormat))(1/1) switched to
> SCHEDULED
> 11/08/2016 21:22:44 DataSource (at main(App.java:25)
> (org.apache.flink.api.java.io.TupleCsvInputFormat))(1/1) switched to
> DEPLOYING
> 11/08/2016 21:22:44 DataSource (at main(App.java:25)
> (org.apache.flink.api.java.io.TupleCsvInputFormat))(1/1) switched to RUNNING  
> 11/08/2016 21:22:44 DataSink (count())(1/1) switched to SCHEDULED
> 11/08/2016 21:22:44 DataSink (count())(1/1) switched to DEPLOYING
> 11/08/2016 21:22:44 DataSink (count())(1/1) switched to RUNNING
> 11/08/2016 21:22:44 DataSink (count())(1/1) switched to FINISHED
> 11/08/2016 21:22:44 DataSource (at main(App.java:25)
> (org.apache.flink.api.java.io.TupleCsvInputFormat))(1/1) switched to
> FINISHED
> 11/08/2016 21:22:44 Job execution switched to status FINISHED.
> Tuple size: 1049866
> Submitting job with JobID: d68d6d775cc222d9fd0728d9666e83de. Waiting for
> job completion.
> Connected to JobManager at Actor[akka.tcp://
> flink@172.19.0.2:6123/user/jobmanager#-658812967]
> 11/08/2016 21:22:45 Job execution switched to status RUNNING.
> 11/08/2016 21:22:45 CHAIN DataSource (at main(App.java:25)
> (org.apache.flink.api.java.io.TupleCsvInputFormat)) -> Map (Map at
> fromTuple2DataSet(Graph.java:343))(1/1) switched to SCHEDULED
>  
> 11/08/2016 21:22:45 CHAIN DataSource (at main(App.java:25)
> (org.apache.flink.api.java.io.TupleCsvInputFormat)) -> Map (Map at
> fromTuple2DataSet(Graph.java:343))(1/1) switched to DEPLOYING
>  
> 11/08/2016 21:22:45 CHAIN DataSource (at main(App.java:25)
> (org.apache.flink.api.java.io.TupleCsvInputFormat)) -> Map (Map at
> fromTuple2DataSet(Graph.java:343))(1/1) switched to RUNNING
> 11/08/2016 21:22:45 CHAIN DataSource (at main(App.java:25)
> (org.apache.flink.api.java.io.TupleCsvInputFormat)) -> Map (Map at
> fromTuple2DataSet(Graph.java:343))(1/1) switched to FINISHED
> 11/08/2016 21:22:45 CHAIN FlatMap (FlatMap at
> fromDataSet(Graph.java:216)) -> Combine(Distinct at
> fromDataSet(Graph.java:216))(1/1) switched to SCHEDULED
> 11/08/2016 21:22:45 CHAIN Map (Map at mapEdges(Graph.java:596)) ->
> FlatMap (FlatMap at getUndirected(Graph.java:926))(1/1) switched to
> SCHEDULED
> 11/08/2016 21:22:45 CHAIN FlatMap (FlatMap at
> fromDataSet(Graph.java:216)) -> Combine(Distinct at
> fromDataSet(Graph.java:216))(1/1) switched to DEPLOYING
> 11/08/2016 21:22:45 CHAIN Map (Map at mapEdges(Graph.java:596)) ->
> FlatMap (FlatMap at getUndirected(Graph.java:926))(1/1) switched to
> DEPLOYING
> 11/08/2016 21:22:45 CHAIN Map (Map at mapEdges(Graph.java:596)) ->
> FlatMap (FlatMap at getUndirected(Graph.java:926))(1/1) switched to RUNNING
> 11/08/2016 21:22:45 CHAIN FlatMap (FlatMap at
> fromDataSet(Graph.java:216)) -> Combine(Distinct at
> fromDataSet(Graph.java:216))(1/1) switched to RUNNING
> 11/08/2016 21:22:45 CoGroup (Messaging)(1/1) switched to SCHEDULED
> 11/08/2016 21:22:45 CoGroup (Messaging)(1/1) switched to DEPLOYING
> 11/08/2016 21:22:45 CoGroup (Messaging)(1/1) switched to RUNNING
> 11/08/2016 21:22:45 CHAIN Reduce (Distinct at
> fromDataSet(Graph.java:216)) -> Map (Map at
> fromDataSet(Graph.java:217))(1/1) switched to SCHEDULED
> 11/08/2016 21:22:45 CHAIN Reduce (Distinct at
> fromDataSet(Graph.java:216)) -> Map (Map at
> fromDataSet(Graph.java:217))(1/1) switched to DEPLOYING
> 11/08/2016 21:22:45 CHAIN Reduce (Distinct at
> fromDataSet(Graph.java:216)) -> Map (Map at
> fromDataSet(Graph.java:217))(1/1) switched to RUNNING
> 11/08/2016 21:22:47 CHAIN Map (Map at mapEdges(Graph.java:596)) ->
> FlatMap (FlatMap at getUndirected(Graph.java:926))(1/1) switched to FINISHED
> 11/08/2016 21:22:47 CHAIN FlatMap (FlatMap at
> fromDataSet(Graph.java:216)) -> Combine(Distinct at
> fromDataSet(Graph.java:216))(1/1) switched to FINISHED
> 11/08/2016 21:22:48 CHAIN Reduce (Distinct at
> fromDataSet(Graph.java:216)) -> Map (Map at
> fromDataSet(Graph.java:217))(1/1) switched to FINISHED
> 11/08/2016 21:22:48 IterationHead(Scatter-gather iteration
> (org.apache.flink.graph.library.ConnectedComponents$CCUpdater@650eab8 |
> org.apache.flink.graph.library.ConnectedComponents$CCMessenger@30f5a68a))(1/1)  
> switched to SCHEDULED
> 11/08/2016 21:22:48 IterationHead(Scatter-gather iteration
> (org.apache.flink.graph.library.ConnectedComponents$CCUpdater@650eab8 |
> org.apache.flink.graph.library.ConnectedComponents$CCMessenger@30f5a68a))(1/1)  
> switched to DEPLOYING
> 11/08/2016 21:22:48 IterationHead(Scatter-gather iteration
> (org.apache.flink.graph.library.ConnectedComponents$CCUpdater@650eab8 |
> org.apache.flink.graph.library.ConnectedComponents$CCMessenger@30f5a68a))(1/1)  
> switched to RUNNING
> 11/08/2016 21:22:48 IterationHead(Scatter-gather iteration
> (org.apache.flink.graph.library.ConnectedComponents$CCUpdater@650eab8 |
> org.apache.flink.graph.library.ConnectedComponents$CCMessenger@30f5a68a))(1/1)  
> switched to FAILED
> java.lang.IllegalArgumentException: Too few memory segments provided. Hash
> Table needs at least 33 memory segments.
> at
> org.apache.flink.runtime.operators.hash.CompactingHashTable.(CompactingHashTable.java:206)
 
> at
> org.apache.flink.runtime.operators.hash.CompactingHashTable.(CompactingHashTable.java:191)
 
> at
> org.apache.flink.runtime.iterative.task.IterationHeadTask.initCompactingHashTable(IterationHeadTask.java:175)
 
> at
> org.apache.flink.runtime.iterative.task.IterationHeadTask.run(IterationHeadTask.java:272)
 
> at
> org.apache.flink.runtime.operators.BatchTask.invoke(BatchTask.java:351)  
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:584)
> at java.lang.Thread.run(Thread.java:745)
>  
> 11/08/2016 21:22:48 Job execution switched to status FAILING.
> java.lang.IllegalArgumentException: Too few memory segments provided. Hash
> Table needs at least 33 memory segments.
> at
> org.apache.flink.runtime.operators.hash.CompactingHashTable.(CompactingHashTable.java:206)
 
> at
> org.apache.flink.runtime.operators.hash.CompactingHashTable.(CompactingHashTable.java:191)
 
> at
> org.apache.flink.runtime.iterative.task.IterationHeadTask.initCompactingHashTable(IterationHeadTask.java:175)
 
> at
> org.apache.flink.runtime.iterative.task.IterationHeadTask.run(IterationHeadTask.java:272)
 
> at
> org.apache.flink.runtime.operators.BatchTask.invoke(BatchTask.java:351)  
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:584)
> at java.lang.Thread.run(Thread.java:745)
>  
> The results I found online so far were not enough, and I am not sure as to
> the best way to solve this.
>  
> If anyone can help diagnose and correct this issue, I would be very
> thankful.
>  
> Best regards,
>  
> Miguel E. Coimbra
> Email: miguel.e.coimbra@gmail.com  
> Skype: miguel.e.coimbra
>  


Mime
View raw message