Yes it's, I'm checking number of columns per line to filter out mailformed *Sent from my ZenFone On Oct 8, 2015 1:19 PM, "Stephan Ewen" <se...@apache.org> wrote:
> There is probably a different CSV input format implementation which drops > invalid lines (too long lines). > > Is that actually desired behavior, simply dropping malformatted input? > > On Thu, Oct 8, 2015 at 7:12 PM, KOSTIANTYN Kudriavtsev < > kudryavtsev.konstan...@gmail.com> wrote: > >> Hm, you was write >> >> I checked all files, one by one and found an issue with a line in one of >> them... It's really unexpected for me as far as I run spark job on the same >> dataset and "wrong" rows were filtered out without issues. >> >> Thanks for help! >> >> Thank you, >> Konstantin Kudryavtsev >> >> On Thu, Oct 8, 2015 at 12:35 PM, Stephan Ewen <se...@apache.org> wrote: >> >>> Ah, that makes sense! >>> >>> The problem is not in the core runtime, it is in the delimited input >>> format. It probably looks for the line split character and never finds it, >>> so that it starts buffering a super large line (gigabytes) which leads to >>> the OOM exception. >>> >>> Can you check whether the line split character and the encoding are >>> properly defined? >>> >>> Would actually be good to define a max line length (sane default + >>> configurable value) that reports when lines seem to extend a maximum length >>> (which is usually a misconfiguration of the split character) >>> >>> Greetings, >>> Stephan >>> >>> >>> On Thu, Oct 8, 2015 at 6:29 PM, KOSTIANTYN Kudriavtsev < >>> kudryavtsev.konstan...@gmail.com> wrote: >>> >>>> 10/08/2015 16:25:48 CHAIN DataSource (at >>>> com.epam.AirSetJobExample$.main(AirSetJobExample.scala:31) >>>> (org.apache.flink.api.java.io.TextInputFormat)) -> Filter (Filter at >>>> com.epam.AirSetJobExample$.main(AirSetJobExample.scala:31)) -> FlatMap >>>> (FlatMap at count(DataSet.scala:523))(1/1) switched to FAILED >>>> java.lang.OutOfMemoryError: Java heap space >>>> at >>>> org.apache.flink.api.common.io.DelimitedInputFormat.readLine(DelimitedInputFormat.java:543) >>>> at >>>> org.apache.flink.api.common.io.DelimitedInputFormat.nextRecord(DelimitedInputFormat.java:453) >>>> at >>>> org.apache.flink.runtime.operators.DataSourceTask.invoke(DataSourceTask.java:176) >>>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:559) >>>> at java.lang.Thread.run(Thread.java:745) >>>> >>>> 10/08/2015 16:25:48 Job execution switched to status FAILING. >>>> 10/08/2015 16:25:48 DataSink >>>> (org.apache.flink.api.java.io.DiscardingOutputFormat@58dbb8cf)(1/1) >>>> switched to CANCELED >>>> 10/08/2015 16:25:48 Job execution switched to status FAILED. >>>> org.apache.flink.client.program.ProgramInvocationException: The program >>>> execution failed: Job execution failed. >>>> at org.apache.flink.client.program.Client.run(Client.java:413) >>>> at org.apache.flink.client.program.Client.run(Client.java:356) >>>> at org.apache.flink.client.program.Client.run(Client.java:349) >>>> at >>>> org.apache.flink.client.program.ContextEnvironment.execute(ContextEnvironment.java:63) >>>> at >>>> org.apache.flink.api.java.ExecutionEnvironment.execute(ExecutionEnvironment.java:789) >>>> at >>>> org.apache.flink.api.scala.ExecutionEnvironment.execute(ExecutionEnvironment.scala:576) >>>> at org.apache.flink.api.scala.DataSet.count(DataSet.scala:524) >>>> at com.epam.AirSetJobExample$.main(AirSetJobExample.scala:35) >>>> at com.epam.AirSetJobExample.main(AirSetJobExample.scala) >>>> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) >>>> at >>>> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) >>>> at >>>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) >>>> at java.lang.reflect.Method.invoke(Method.java:606) >>>> at >>>> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:437) >>>> at >>>> org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:353) >>>> at org.apache.flink.client.program.Client.run(Client.java:315) >>>> at >>>> org.apache.flink.client.CliFrontend.executeProgram(CliFrontend.java:582) >>>> at org.apache.flink.client.CliFrontend.run(CliFrontend.java:288) >>>> at >>>> org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:878) >>>> at >>>> org.apache.flink.client.CliFrontend.main(CliFrontend.java:920) >>>> Caused by: org.apache.flink.runtime.client.JobExecutionException: Job >>>> execution failed. >>>> at >>>> org.apache.flink.runtime.jobmanager.JobManager$$anonfun$receiveWithLogMessages$1.applyOrElse(JobManager.scala:314) >>>> at >>>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33) >>>> at >>>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33) >>>> at >>>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25) >>>> at >>>> org.apache.flink.yarn.ApplicationMasterActor$$anonfun$receiveYarnMessages$1.applyOrElse(ApplicationMasterActor.scala:100) >>>> at scala.PartialFunction$OrElse.apply(PartialFunction.scala:162) >>>> at >>>> org.apache.flink.runtime.ActorLogMessages$$anon$1.apply(ActorLogMessages.scala:36) >>>> at >>>> org.apache.flink.runtime.ActorLogMessages$$anon$1.apply(ActorLogMessages.scala:29) >>>> at >>>> scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118) >>>> at >>>> org.apache.flink.runtime.ActorLogMessages$$anon$1.applyOrElse(ActorLogMessages.scala:29) >>>> at akka.actor.Actor$class.aroundReceive(Actor.scala:465) >>>> at >>>> org.apache.flink.runtime.jobmanager.JobManager.aroundReceive(JobManager.scala:92) >>>> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516) >>>> at akka.actor.ActorCell.invoke(ActorCell.scala:487) >>>> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254) >>>> at akka.dispatch.Mailbox.run(Mailbox.scala:221) >>>> at akka.dispatch.Mailbox.exec(Mailbox.scala:231) >>>> at >>>> scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) >>>> at >>>> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) >>>> at >>>> scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) >>>> at >>>> scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) >>>> Caused by: java.lang.OutOfMemoryError: Java heap space >>>> at >>>> org.apache.flink.api.common.io.DelimitedInputFormat.readLine(DelimitedInputFormat.java:543) >>>> at >>>> org.apache.flink.api.common.io.DelimitedInputFormat.nextRecord(DelimitedInputFormat.java:453) >>>> at >>>> org.apache.flink.runtime.operators.DataSourceTask.invoke(DataSourceTask.java:176) >>>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:559) >>>> at java.lang.Thread.run(Thread.java:745) >>>> >>>> >>>> Thank you, >>>> Konstantin Kudryavtsev >>>> >>>> On Thu, Oct 8, 2015 at 12:23 PM, Stephan Ewen <se...@apache.org> wrote: >>>> >>>>> Can you paste the exception stack trace? >>>>> >>>>> On Thu, Oct 8, 2015 at 6:15 PM, KOSTIANTYN Kudriavtsev < >>>>> kudryavtsev.konstan...@gmail.com> wrote: >>>>> >>>>>> It's DataSet program that performs simple filtering, crossjoin and >>>>>> aggregation. >>>>>> >>>>>> I'm using Hadoop S3 FileSystem (not Emr) as far as Flink's s3 >>>>>> connector doesn't work at all. >>>>>> >>>>>> Currently I have 3 taskmanagers each 5k MB, but I tried different >>>>>> configurations and all leads to the same exception >>>>>> >>>>>> *Sent from my ZenFone >>>>>> On Oct 8, 2015 12:05 PM, "Stephan Ewen" <se...@apache.org> wrote: >>>>>> >>>>>>> Can you give us a bit more background? What exactly is your program >>>>>>> doing? >>>>>>> >>>>>>> - Are you running a DataSet program, or a DataStream program? >>>>>>> - Is it one simple source that reads from S3, or are there >>>>>>> multiple sources? >>>>>>> - What operations do you apply on the CSV file? >>>>>>> - Are you using Flink's S3 connector, or the Hadoop S3 file system? >>>>>>> >>>>>>> Greetings, >>>>>>> Stephan >>>>>>> >>>>>>> >>>>>>> On Thu, Oct 8, 2015 at 5:58 PM, KOSTIANTYN Kudriavtsev < >>>>>>> kudryavtsev.konstan...@gmail.com> wrote: >>>>>>> >>>>>>>> Hi guys, >>>>>>>> >>>>>>>> I'm running FLink on EMR with 2 m3.xlarge (each 16 GB RAM) and >>>>>>>> trying to process 3.8 GB CSV data from S3. I'm surprised the fact that >>>>>>>> Flink failed with OutOfMemory: Java Heap space >>>>>>>> >>>>>>>> I tried to find the reason: >>>>>>>> 1) to identify TaskManager with a command ps aux | grep TaskManager >>>>>>>> 2) then build Heap histo: >>>>>>>> $ jmap -histo:live 19648 | head -n23 >>>>>>>> num #instances #bytes class name >>>>>>>> ---------------------------------------------- >>>>>>>> 1: 131018 3763501304 [B >>>>>>>> 2: 61022 7820352 <methodKlass> >>>>>>>> 3: 61022 7688456 <constMethodKlass> >>>>>>>> 4: 4971 5454408 <constantPoolKlass> >>>>>>>> 5: 4966 4582232 <instanceKlassKlass> >>>>>>>> 6: 4169 3003104 <constantPoolCacheKlass> >>>>>>>> 7: 15696 1447168 [C >>>>>>>> 8: 1291 638824 [Ljava.lang.Object; >>>>>>>> 9: 5318 506000 java.lang.Class >>>>>>>> >>>>>>>> >>>>>>>> Do you have any ideas what can be the reason and how it can be >>>>>>>> fixed? >>>>>>>> Is Flink uses out-of-heap memory? >>>>>>>> >>>>>>>> >>>>>>>> Thank you, >>>>>>>> Konstantin Kudryavtsev >>>>>>>> >>>>>>> >>>>>>> >>>>> >>>> >>> >> >
