Spark dataframe - анализ журналов ошибок журналов приложений

Question

У меня есть большая потребность в данных для анализа ошибки исключения веб-приложения. Мой подход выглядит следующим образом:

1. Поток в журнал ошибок приложения weblogic в hadoop с использованием флюма или других потоковых инструментов.
2. Загрузите данные в Spark Dataframe.
3. записи Спарк SQL запросов для анализа данных об ошибках

У нас есть таблица журнала ошибок БД. Я буду использовать это как другой источник данных для корреляции исключений БД веб-логики. Данные ошибки weblogic - это формат CSV, разделенный двумя символами канала («||»). Но проблема с входными данными заключается в том, что последние данные столбца распространяются на несколько строк, как показано ниже. Spark рассматривает продолжение последнего столбца на следующей строке как «новую строку», и, следовательно, загрузка становится неудачной. Цените, если кто-нибудь подумает о том, как решить эту проблему.

|| 20160704 || 01: 58: 32294 || 396c0a8e2470e7a21467611910768 || com.seic.dataservices.impl.InstrumentSearchDoImpl || [АКТИВНЫЙ] ExecuteThread: '9' для очереди: 0) .... INSTRUMENT_ID (1004915) PRICE_DATE (01-JUL-16) не существует в таблице TABEL_NAME. Данные о ценах не найден .. ORA-06512: на "Qxx_xxx.ERROR_PKG", строка 502 ORA-06512: на "Qxx_xxx.IM_PRICING", строка 6221 ORA-06512: в строке 1)

-UK

Обновление: отредактирован входной набор данных.

|| 20160704 || 00: 32: 48544 || c0a07f3289f452801467606768492 || com.seic.dataservices.impl.GetInstDetailsForMaintImpl || [АКТИВНЫЙ] ExecuteThread: '12' для очереди: «weblogic.kernel.Default (само- tuning) '|| ERROR || ExceptionFactoryMsg: com.seic.dataservices.lib.DataServiceSqlException - Ошибка - при обработке этого запроса возникло исключение SQL. - EX4 - - q02_Desktop_MS1 # 20160704003248544 # 4 - с дополнительной информацией: (код ошибки: 6550 -) - вызвано (java.sql.SQLException: ORA-06550: строка 1, столбец 25: PLS-00302: компонент 'GET_ASSET_TEMPLATE' должен быть объявлен ORA-06550: строка 1, столбец 7: PL/SQL: заявление игнорируется ) com.seic.dataservices.lib.DataServiceSqlException - Ошибка. При обработке этого запроса возникло исключение SQL. - EX4 - - q02_Desktop_MS1 # 20160704003248544 # 4 - с дополнительной информацией: (код ошибки: 6550 -) - вызвано (java.sql.SQLException: ORA-06550: строка 1, столбец 25: PLS-00302: компонент 'GET_ASSET_TEMPLATE' должен быть объявлен ОР-06550: строка 1, столбец 7: PL/SQL: Заявление игнорируется ) || 20160704 || 00: 32: 48551 || c0a07f3289f452801467606768492 || com.seic.common.presentation.exception.SeiExceptionHandler || [ACTIVE] ExecuteThread: '12' для очереди: 'weblogic.kernel.Default (self-tuning)' || ERROR || Non-SeiException javax.servlet.ServletException: ошибка в базовом классе DesktopAction (исключение) at com.seic.common.presentation.action.DesktopAction.execute (DesktopAction.java:368) по адресу org.apache.struts.chain.commands.servlet.ExecuteAction.execute (ExecuteAction.jav a: 58) на org.apache.struts.chain.commands.AbstractExecuteAction.execute (AbstractExecuteAction.java:67) на org.apache.struts.chain.commands.ActionCommandBase.execute (ActionCommandBase.java:51) at org.apache.commons.chain.impl.ChainBase.execute (ChainBase.java:191) на org.apache.commons.chain.generic.LookupCommand.execute (LookupCommand.java:305) на org.apache.commons. chain.impl.ChainBase.execute (ChainBase.java:191)

Answer 1

Хорошо, я дам вам представление о том, что вы можете сделать.

Предположим, у вас всегда есть пара рядов, даже не в порядке. Попробуйте сделать это:

rdd = sc.textFile('file.log').zipWithIndex() 
rddFirsts = rdd.filter(lambda x: not(x[1] % 2)).map(lambda x: (x[1], x[0])) 
rddSeconds = rdd.filter(lambda x: x[1] % 2).map(lambda x: (x[1]-1, x[0])) 
rdd = rddFirsts.join(rddSeconds) 

С помощью этих команд он будет работать должным образом, как вы хотите. Я знаю, что это может занять много времени. Но он отлично работает.

Я тестировал в Спарк 1.5.2

EDITED

---

Для Скале:

val rdd = sc.textFile("file.log").zipWithIndex() 
val rddFirsts = rdd.filter(x => (x._2 % 2) == 0).map(x => (x._2, x._1)) 
val rddSeconds = rdd.filter(x => (x._2 % 2) != 0).map(x => (x._2-1, x._1)) 
val NewRdd = rddFirsts.join(rddSecons)