Reading CSVs with new lines in fields with Spark

Question

I was trying to load the below weblogic domain log (application error log) into Spark dataframe. I created a RDD and converted the RDD into dataframe. I was able to load the data successfully for the first two rows because the records are not spread over to multiple lines. But, for the third row (highlighted in bold), the record is spread over multiple lines and Spark assumes the continuation of the last field on the next line as new record. Can anybody suggest any workaround so that the multi-line data can be considered as part of a single record? Any pointers would be really helpful.

Scala commands:

case class error(error_log_id: String,procedure_name: String,section_name: String,sei_user_id: String,error_timestamp: String,error_number: String,error_message: String,error_stack: String,error_backtrace: String,java_thread_id: String) 
val errorLog = sc.textFile("errorLog.csv").map(_.split("#")).map(p => error(p(0),p(1),p(2),p(3),p(4),p(5),p(6),p(7),p(8),p(9))).toDF() 
errorLog.registerTempTable("error_log")

Weblogic error log:

||20160704||00:30:50,595||4affd321959f27e41467606650584||com.seic.common.presentation.controller.ProcessControllerAction||[ACTIVE] ExecuteThread: '7' for queue: 'weblogic.kernel.Default (self-tuning)'||INFO ||Entered performPreCondition for event=evtLocationContext event = evtLocationContext 
||20160704||00:30:50,607||4affd321959f27e41467606650584||com.seic.desktop.relationshipmgmt.presentation.helper.ContextTreeHelper||[ACTIVE] ExecuteThread: '7' for queue: 'weblogic.kernel.Default (self-tuning)'||INFO ||ContextEntityWrapperTO found in Session 
**||20160704||01:58:32,182||d3ac1cac46b457251467611729203||com.seic.common.presentation.filter.JSPTrackerFilter||[ACTIVE] ExecuteThread: '2' for queue: 'weblogic.kernel.Default (self-tuning)'||INFO ||JSPTrackerFilter  :    User Id = 8000551,    User Firm Id = 1, JSP Name = /SEIDesktop/jsp/common/maintainSession.jsp    , Date & Timestamp = 7/4/16 1:58 AM    ,Browser Name & Version = Mozilla/5.0 ||20160704||01:58:32,294||396c0a8e2470e7a21467611910768||com.seic.dataservices.impl.InstrumentSearchDoImpl||[ACTIVE] ExecuteThread: '9' for queue: 'weblogic.kernel.Default (self-tuning)'||ERROR||ExceptionFactoryMsg:  com.seic.dataservices.lib.DataServiceSqlException - Error - A SQL exception was encountered while processing this request. - EX13 -  - q02_Desktop_MS1#20160704015832294#13 - With Additional Info:  (Error Code:  20010 -  ) - Caused By (java.sql.SQLException: ORA-20010: PO_ID(1000) INSTRUMENT_ID(1004915) PRICE_DATE(01-JUL-16) does not exist in Table PO_SECURITY_PRICE.  Price data not found.. 
ORA-06512: at "xxx_xxx.ERROR_PKG", line 502 
ORA-06512: at "xxx_xxx.xx_PRICING", line 6221 
ORA-06512: at line 1 
) 
com.seic.dataservices.lib.DataServiceSqlException - Error - A SQL exception was encountered while processing this request. - EX13 -  - q02_Desktop_MS1#20160704015832294#13 - With Additional Info:  (Error Code:  20010 -  ) - Caused By (java.sql.SQLException: ORA-20010: PO_ID(1000) INSTRUMENT_ID(1004915) PRICE_DATE(01-JUL-16) does not exist in Table PO_SECURITY_PRICE.  Price data not found.. 
ORA-06512: at "xxx_xxx.ERROR_PKG", line 502 
ORA-06512: at "xxx_xxx.xx_PRICING", line 6221 
ORA-06512: at line 1 
) 
                at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) 
                at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:39) 
                at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:27) 
                at java.lang.reflect.Constructor.newInstance(Constructor.java:513) 
                at com.seic.framework.exceptions.ExceptionFactory.getSeiException(ExceptionFactory.java:536) 
                at com.seic.framework.exceptions.ExceptionFactory.createException(ExceptionFactory.java:269) 
                at com.seic.framework.exceptions.ExceptionFactory.createException(ExceptionFactory.java:172) 
                at com.seic.dataservices.dm.InstrumentSearchDo.Get(InstrumentSearchDo.java:3796) 
                at com.seic.dataservices.impl.InstrumentSearchDoPkFilterHelper.doGet(InstrumentSearchDoPkFilterHelper.java:401) 
                at com.seic.dataservices.lib.DefaultValuesDataServiceHelper.doGet(DefaultValuesDataServiceHelper.java:34) 
                at com.seic.dataservices.lib.DataServiceImpl.getList(DataServiceImpl.java:81) 
                at com.seic.dataservices.lib.DataServiceImpl.get(DataServiceImpl.java:127) 
                at com.seic.dataservices.lib.DataServiceImpl.get(DataServiceImpl.java:144) 
                at com.seic.ads.setup.SearchServices.searchInstruments(SearchServices.java:213) 
                at com.seic.ads.ejb.ADSServicesEjbBean.searchInstruments(ADSServicesEjbBean.java:699) 
                at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 
                at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39) 
                at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25) 
                at java.lang.reflect.Method.invoke(Method.java:597) 
                at com.bea.core.repackaged.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:310) 
                at com.bea.core.repackaged.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:182) 
                at com.bea.core.repackaged.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:149) 
                at com.bea.core.repackaged.springframework.aop.support.DelegatingIntroductionInterceptor.doProceed(DelegatingIntroductionInterceptor.java:131)
                at com.bea.core.repackaged.springframework.aop.support.DelegatingIntroductionInterceptor.invoke(DelegatingIntroductionInterceptor.java:119) 
                at com.bea.core.repackaged.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:171) 
                at com.bea.core.repackaged.springframework.jee.spi.MethodInvocationVisitorImpl.visit(MethodInvocationVisitorImpl.java:37) 
                at weblogic.ejb.container.injection.EnvironmentInterceptorCallbackImpl.callback(EnvironmentInterceptorCallbackImpl.java:54) 
                at com.bea.core.repackaged.springframework.jee.spi.EnvironmentInterceptor.invoke(EnvironmentInterceptor.java:50) 
                at com.bea.core.repackaged.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:171) 
                at com.bea.core.repackaged.springframework.aop.interceptor.ExposeInvocationInterceptor.invoke(ExposeInvocationInterceptor.java:89)
                at com.bea.core.repackaged.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:171) 
                at com.bea.core.repackaged.springframework.aop.support.DelegatingIntroductionInterceptor.doProceed(DelegatingIntroductionInterceptor.java:131)
                at com.bea.core.repackaged.springframework.aop.support.DelegatingIntroductionInterceptor.invoke(DelegatingIntroductionInterceptor.java:119) 
                at com.bea.core.repackaged.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:171) 
                at com.bea.core.repackaged.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:204) 
                at com.sun.proxy.$Proxy2528.searchInstruments(Unknown Source) 
                at com.seic.ads.ejb.ADSServicesEjbBean_otv51e_ADSServicesEjbImpl.searchInstruments(ADSServicesEjbBean_otv51e_ADSServicesEjbImpl.java:2975)
                at com.seic.ads.ejb.ADSServicesEjbBean_otv51e_ADSServicesEjbImpl_WLSkel.invoke(Unknown Source) 
                at weblogic.rmi.internal.ServerRequest.sendReceive(ServerRequest.java:174) 
                at weblogic.rmi.cluster.ClusterableRemoteRef.invoke(ClusterableRemoteRef.java:345) 
                at weblogic.rmi.cluster.ClusterableRemoteRef.invoke(ClusterableRemoteRef.java:259) 
                at com.seic.ads.ejb.ADSServicesEjbBean_otv51e_ADSServicesEjbImpl_1033_WLStub.searchInstruments(Unknown Source) 
                at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 
                at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39) 
                at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25) 
                at java.lang.reflect.Method.invoke(Method.java:597) 
                at weblogic.ejb.container.internal.RemoteBusinessIntfProxy.invoke(RemoteBusinessIntfProxy.java:84) 
                at com.sun.proxy.$Proxy2527.searchInstruments(Unknown Source) 
                at com.seic.ads.delegate.ADSServicesDelegate.searchInstruments(ADSServicesDelegate.java:490) 
                at com.seic.desktop.ads.presentation.helper.AssetSearchHelper.getInstruments(AssetSearchHelper.java:423) 
                at com.seic.desktop.ads.presentation.action.AssetSearchAction.handleSearchCriteriaEvents(AssetSearchAction.java:488) 
                at com.seic.desktop.ads.presentation.action.AssetSearchAction.performPostCondition(AssetSearchAction.java:254) 
                at com.seic.common.presentation.controller.ProcessControllerAction.handleEvents(ProcessControllerAction.java:78) 
                at com.seic.common.presentation.controller.ProcessControllerActionBase.process(ProcessControllerActionBase.java:119) 
                at com.seic.common.presentation.action.DesktopAction.execute(DesktopAction.java:357) 
                at org.apache.struts.chain.commands.servlet.ExecuteAction.execute(ExecuteAction.java:58) 
                at org.apache.struts.chain.commands.AbstractExecuteAction.execute(AbstractExecuteAction.java:67) 
                at org.apache.struts.chain.commands.ActionCommandBase.execute(ActionCommandBase.java:51) 
                at org.apache.commons.chain.impl.ChainBase.execute(ChainBase.java:191) 
                at org.apache.commons.chain.generic.LookupCommand.execute(LookupCommand.java:305) 
                at org.apache.commons.chain.impl.ChainBase.execute(ChainBase.java:191) 
                at org.apache.struts.chain.ComposableRequestProcessor.process(ComposableRequestProcessor.java:283) 
                at org.apache.struts.action.ActionServlet.process(ActionServlet.java:1913) 
                at org.apache.struts.action.ActionServlet.doPost(ActionServlet.java:462) 
                at javax.servlet.http.HttpServlet.service(HttpServlet.java:727) 
                at javax.servlet.http.HttpServlet.service(HttpServlet.java:820) 
                at weblogic.servlet.internal.StubSecurityHelper$ServletServiceAction.run(StubSecurityHelper.java:227) 
                at weblogic.servlet.internal.StubSecurityHelper.invokeServlet(StubSecurityHelper.java:125) 
                at weblogic.servlet.internal.ServletStubImpl.execute(ServletStubImpl.java:300) 
                at weblogic.servlet.internal.TailFilter.doFilter(TailFilter.java:26) 
                at weblogic.servlet.internal.FilterChainImpl.doFilter(FilterChainImpl.java:56) 
                at com.seic.common.presentation.filter.UserContextFilter.doFilter(UserContextFilter.java:65) 
                at weblogic.servlet.internal.FilterChainImpl.doFilter(FilterChainImpl.java:56) 
                at com.seic.common.presentation.helper.InjectionAttackFilter.chainNext(InjectionAttackFilter.java:194) 
                at com.seic.common.presentation.helper.InjectionAttackFilter.doFilter(InjectionAttackFilter.java:188) 
                at weblogic.servlet.internal.FilterChainImpl.doFilter(FilterChainImpl.java:56) 
                at weblogic.servlet.internal.WebAppServletContext$ServletInvocationAction.doIt(WebAppServletContext.java:3684) 
                at weblogic.servlet.internal.WebAppServletContext$ServletInvocationAction.run(WebAppServletContext.java:3650) 
                at weblogic.security.acl.internal.AuthenticatedSubject.doAs(AuthenticatedSubject.java:321) 
                at weblogic.security.service.SecurityManager.runAs(SecurityManager.java:121) 
                at weblogic.servlet.internal.WebAppServletContext.securedExecute(WebAppServletContext.java:2268) 
                at weblogic.servlet.internal.WebAppServletContext.execute(WebAppServletContext.java:2174) 
                at weblogic.servlet.internal.ServletRequestImpl.run(ServletRequestImpl.java:1446) 
                at weblogic.work.ExecuteThread.execute(ExecuteThread.java:201) 
                at weblogic.work.ExecuteThread.run(ExecuteThread.java:173) 
Caused by: java.sql.SQLException: ORA-20010: PO_ID(222) INSTRUMENT_ID(1111112) PRICE_DATE(01-JUL-16) does not exist in Table xxx.  Price data not found.. 
ORA-06512: at "Q02_EDB.ERROR_PKG", line 502 
ORA-06512: at "Q02_EDB.IM_PRICING", line 6221 
ORA-06512: at line 1 
                at oracle.jdbc.driver.T4CTTIoer.processError(T4CTTIoer.java:440) 
                at oracle.jdbc.driver.T4CTTIoer.processError(T4CTTIoer.java:396) 
                at oracle.jdbc.driver.T4C8Oall.processError(T4C8Oall.java:837) 
                at oracle.jdbc.driver.T4CTTIfun.receive(T4CTTIfun.java:445) 
                at oracle.jdbc.driver.T4CTTIfun.doRPC(T4CTTIfun.java:191) 
                at oracle.jdbc.driver.T4C8Oall.doOALL(T4C8Oall.java:523) 
                at oracle.jdbc.driver.T4CStatement.doOall8(T4CStatement.java:193) 
                at oracle.jdbc.driver.T4CStatement.fetch(T4CStatement.java:1073) 
                at oracle.jdbc.driver.OracleResultSetImpl.close_or_fetch_from_next(OracleResultSetImpl.java:359) 
                at oracle.jdbc.driver.OracleResultSetImpl.next(OracleResultSetImpl.java:263) 
                at weblogic.jdbc.wrapper.ResultSet_oracle_jdbc_driver_OracleResultSetImpl.next(Unknown Source) 
                at com.seic.dataservices.dm.InstrumentSearchDo.Get(InstrumentSearchDo.java:3551) 
                ... 77 more 
||20160704||01:58:32,294||396c0a8e2470e7a21467611910768||com.seic.dataservices.impl.InstrumentSearchDoImpl||[ACTIVE] ExecuteThread: '9' for queue: 'weblogic.kernel.Default (self-tuning)'||ERROR||ExceptionFactoryMsg:  com.seic.dataservices.lib.DataServiceSqlException - Error - A SQL exception was encountered while processing this request. - EX13 -  - q02_Desktop_MS1#20160704015832294#13 - With Additional Info:  (Error Code:  20010 -  ) - Caused By (java.sql.SQLException: ORA-20010: PO_ID(1000) INSTRUMENT_ID(1004915) PRICE_DATE(01-JUL-16) does not exist in Table TABEL_NAME.  Price data not found.. 
ORA-06512: at "Qxx_xxx.ERROR_PKG", line 502 
ORA-06512: at "Qxx_xxx.IM_PRICING", line 6221 
ORA-06512: at line 1 
)**

I may be wrong, but using line breaks in something that is meant to be CSV-parseable, without escaping the multi-line column value in quotes, seems to break the expectations of most CSV parsers. This looks like some special format as well, as indicated by the double-asterisk at the start of that multi-line row (and the inconsistent trailing double-asterisk later) -- which will definitely break all CSV parsers. You may have better luck forking the code for one of the parser libraries, tweaking it to better fit this alternate format, then using it in a similar way as the answer proposed below. — Brian Cline, Aug 7 '16 at 23:18

Aneel · Accepted Answer · 2017-07-12 22:21:42Z

5

There is a recently-added feature in Spark 2.2.0...

spark.read.csv(file, multiLine=True)

https://issues.apache.org/jira/browse/SPARK-19610

https://issues.apache.org/jira/browse/SPARK-20980

answered Jul 12 '17 at 22:21

Aneel

1511 silver badge4 bronze badges

add a comment |

Stephen Rauch · Accepted Answer · 2018-04-15 02:27:25Z

2

Augmenting Aneel's answer, I had to add escape='"' option get this working properly. Spark 2.3

spark.read.csv(DATA_FILE, sep=',', escape='"', header=True, 
               inferSchema=True, multiLine=True).count()

159571

Interestingly, Pandas can read this without any additional instructions.

pd.read_csv(DATA_FILE).shape

(159571, 8)

edited Apr 15 '18 at 2:27

Stephen Rauch♦

1,6699 gold badges16 silver badges31 bronze badges

answered Apr 15 '18 at 2:07

Thusitha

811 silver badge5 bronze badges

add a comment |

Emre · Accepted Answer · 2016-07-14 04:34:17Z

1

Try setting parserLib to univocity:

> val df = sqlContext.read.options(Map("parserLib" -> "univocity", "header" -> "true")).format("csv").load("test.csv")
> df.collect.foreach(println)

[hello
world,hi mom,and
dad]
[1,2
more,3]
[foo,bar,just
one]

test.csv is as follows:

a,b,c
"hello
world",hi mom,"and
dad"
1,"2
more",3
foo,bar,"just
one"

Alternatively, replace the new line characters with something else before parsing, and back again afterwards.

edited Jul 14 '16 at 4:34

answered Jul 12 '16 at 17:55

Emre

9,6911 gold badge25 silver badges37 bronze badges

$\begingroup$ Thanks. I am trying for a real time or near real-time solution. The log files is huge. So, if I need to replace the new line character for a large log file, that will be a time consuming operation. $\endgroup$ – uk2016 Jul 12 '16 at 19:02
$\begingroup$ The univocity parser is probably assuming those columns are properly escaped with double-quotes. If creating your own parser library based on code for one of the existing CSV ones isn't an option, and preprocessing it isn't either, there's really no clean way around this. What happens if an input split boundary falls right through the middle of one of these multi-line messages? $\endgroup$ – Brian Cline Aug 7 '16 at 23:30

add a comment |

Stack Exchange Network

current community

your communities

more stack exchange communities

Reading CSVs with new lines in fields with Spark

3 Answers 3

Your Answer

Not the answer you're looking for? Browse other questions tagged apache-spark csv or ask your own question.

Hot Network Questions

Reading CSVs with new lines in fields with Spark

3 Answers 3

Your Answer

Sign up or log in

Post as a guest

Not the answer you're looking for? Browse other questions tagged apache-spark csv or ask your own question.

Related

Hot Network Questions