6
$\begingroup$

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 
)**
Improve this question
$\endgroup$
1
  • 1
    $\begingroup$ 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. $\endgroup$ – Brian Cline Aug 7 '16 at 23:18
5
$\begingroup$

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

Improve this answer
$\endgroup$
2
$\begingroup$

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)
Improve this answer
$\endgroup$
1
$\begingroup$

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.

Improve this answer
$\endgroup$
2
  • $\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

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

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