使用 Spark 在字段中读取带有新行的 CSV

数据挖掘 阿帕奇火花 CSV
2021-10-10 18:25:48

我试图将以下 weblogic 域日志（应用程序错误日志）加载到 Spark 数据帧中。我创建了一个 RDD 并将 RDD 转换为数据框。我能够成功加载前两行的数据，因为记录没有分布到多行。但是，对于第三行（以粗体突出显示），记录分布在多行中，Spark 假定下一行的最后一个字段的延续作为新记录。任何人都可以提出任何解决方法，以便可以将多行数据视为单个记录的一部分吗？任何指针都会非常有帮助。

斯卡拉命令： 

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 错误日志： 

||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 
)**
3个回答

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

增强 Aneel 的回答，我不得不添加escape='"'选项让它正常工作。火花 2.3

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

159571

有趣的是，Pandas 无需任何额外说明即可阅读此内容。 

pd.read_csv(DATA_FILE).shape

(159571, 8)

尝试设置parserLib为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如下：

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

或者，在解析之前用其他内容替换新行字符，然后再返回。

其它你可能感兴趣的问题
上一篇什么是强化学习？ 下一篇是否有已知的卷积网络架构来计算图像的对象掩码？