Как преобразовать sqlite json column в строку или структуру Spark?

Question

У меня есть база данных sqlite со столбцами, сохраненными как json, некоторые являются просто массивами, а некоторые массивом объектов Данные не слишком большие, около 1 миллиона строк в таблице и еще 6 миллионов в другой таблице. Теперь я хотел бы повысить скорость запросов и извлечь эти данные во что-то индексированное и более управляемое

Проблема в том, что spark обрабатывает столбцы json как BigDecimal, и я не знаю, почему или как это решить, нашел некоторые вещи, но ничего не помогло.

Caused by: java.sql.SQLException: Bad value for type BigDecimal : [56641575300, 56640640900, 56640564100, 56640349700, 18635841800, 54913035400, 6505719940, 56641287800, 7102147726, 57202227222, 57191928343, 18633330200, 57193578904, 7409778074, 7409730079, 55740247200, 56641355300, 18635857700, 57191972388, 54912606500, 6601960745, 57191972907, 56641923500, 56640256300, 54911965100, 45661930800, 55474245300, 7409541556, 7409694518, 56641363000, 56519446200, 6504106170, 57191975866, 56640736700, 55463741500, 56640319300, 56640861000, 54911965000, 56561401800, 6504731849, 24342836300, 7402491855, 22950414800, 6507741522, 6504199636, 7102381436, 57191895642, 18634536800, 57196623329, 7005988322, 56013334500, 18634278500, 57191983462, 7409545828, 57204194408, 56641031400, 56641436400, 6504659572, 36829162100, 24766932600, 8256434300]
    at org.sqlite.jdbc3.JDBC3ResultSet.getBigDecimal(JDBC3ResultSet.java:196)

Я пытался загрузить драйвер sqlite, а затем открыть базу данных с помощью SQLContext

.

df = sqlContext.read.format('jdbc').options(url='jdbc:sqlite:../cache/iconic.db', dbtable='coauthors', driver='org.sqlite.JDBC').load()

После того, как спарк пожаловался на тип столбца, я попытался привести его к типу строки, чтобы он мог быть затем проанализирован как json

schema = ArrayType(IntegerType())
df.withColumn('co_list', from_json(df['co_list'].cast(StringType()), schema))

Но эта ошибка выдает ту же ошибку, поскольку ничего не изменила

Также я попытался установить схему таблицы с самого начала, но похоже, что pyspark не позволяет мне сделать это

df = sqlContext.read.schema([...]).format('jdbc')...
# Throws
pyspark.sql.utils.AnalysisException: 'jdbc does not allow user-specified schemas.;'

Строки выглядят так

# First table
1   "[{""surname"": ...}]"  "[[{""frequency"": ""58123"", ...}]]"   74072   14586   null    null    null    "{""affiliation-url"":}"    "[""SOCI""]"    null    0   0   1

# Second table
505 "[{""surname"": ""Blondel"" ...}, {""surname"": ""B\u0153ge"" ..}, ...]"    "1999-12-01"    21  null    null    null    0

Надеюсь, что есть способ.

Answer 1

Нашли решение, база данных должна быть загружена с помощью jdbc reader и для настройки приведения столбцов необходимо передать свойство драйверу

Вот решение

connectionProperties = { 
    "customSchema": 'id INT, co_list STRING, last_page INT, saved INT',
    "driver": 'org.sqlite.JDBC'
}

df = sqlContext.read.jdbc(url='jdbc:sqlite:../cache/iconic.db', table='coauthors', properties=connectionProperties)

Таким образом, у вас есть контроль над внутренним сопоставлением столбцов таблицы базы данных.