Spark / Neo4j выдает ошибку: RuntimeException: java.util.Collections $ UnmodifiableRandomAccessList не является допустимым внешним типом для схемы строки

Точный запрос:

call spatial.bbox('geom', {lat:37.5,lon:43.4}, {lat:37.6,lon:43.5}) yield node return node.altitude as altitude, node.detect_type as detect_type, node.gtype as gtype, node.toDateFormatLong as toDateFormatLong, node.change_area as change_area, node.latitude as latitude, node.longitude as longitude, node.fromDateFormatLong as fromDateFormatLong, node.iids as iids, node.detect_strength as detect_strength, node.fromDate as fromDate, node.bbox as bbox ORDER BY node.toDateFormatLong DESC

Пример набора данных:

╒══════════╤═════════════╤═══════╤══════════════════╤═════════════╤══════════╤═══════════╤════════════════════╤═════════════════════════════════════════════════════════════════════╤═════════════════╤══════════╤═════════════════════════════════════════════╕
│"altitude"│"detect_type"│"gtype"│"toDateFormatLong"│"change_area"│"latitude"│"longitude"│"fromDateFormatLong"│"iids"                                                               │"detect_strength"│"fromDate"│"bbox"                                       │
╞══════════╪═════════════╪═══════╪══════════════════╪═════════════╪══════════╪═══════════╪════════════════════╪═════════════════════════════════════════════════════════════════════╪═════════════════╪══════════╪═════════════════════════════════════════════╡
│-1        │"Arrival"    │1      │20161104          │16981        │37.5608649│43.4297988 │20161023            │"23OCT16S1A89377_09_IW1_09_pp_1231_04NOV16S1A90776_09_123_31_TT_QQQQ”│7.2              │"23OCT16" │[43.4297988,37.5608649,43.4297988,37.5608649]│
├──────────┼─────────────┼───────┼──────────────────┼─────────────┼──────────┼───────────┼────────────────────┼─────────────────────────────────────────────────────────────────────┼─────────────────┼──────────┼─────────────────────────────────────────────┤
│-1        │"Arrival"    │1      │20161104          │3123         │37.56749  │43.4807208 │20161023            │"23OCT16S1A89377_09_IW1_09_pp_1231_04NOV16S1A90776_09_124_32_TT_QQQQ"│7.5              │"23OCT16" │[43.4807208,37.56749,43.4807208,37.56749]    │
├──────────┼─────────────┼───────┼──────────────────┼─────────────┼──────────┼───────────┼────────────────────┼─────────────────────────────────────────────────────────────────────┼─────────────────┼──────────┼─────────────────────────────────────────────┤

что я звоню

try {
 val initialDf2 = neo.cypher(query).loadDataFrame
 val someVal = initialDf2.collectAsList()
} catch {
   case e: Exception => e.printStackTrace
}

Я получаю такую ​​ошибку:

 17/09/18 08:44:48 ERROR TaskSetManager: Task 0 in stage 298.0 failed 1 times; aborting job
    org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 298.0 failed 1 times, most recent failure: Lost task 0.0 in stage 298.0 (TID 298, localhost, executor driver): java.lang.RuntimeException: Error while encoding: java.lang.RuntimeException: java.util.Collections$UnmodifiableRandomAccessList is not a valid external type for schema of string
    if (assertnotnull(input[0, org.apache.spark.sql.Row, true]).isNullAt) null else validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 0, altitude), DoubleType) AS altitude#1678
    if (assertnotnull(input[0, org.apache.spark.sql.Row, true]).isNullAt) null else staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 1, detect_type), StringType), true) AS detect_type#1679
    if (assertnotnull(input[0, org.apache.spark.sql.Row, true]).isNullAt) null else validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 2, gtype), LongType) AS gtype#1680L
    if (assertnotnull(input[0, org.apache.spark.sql.Row, true]).isNullAt) null else validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 3, toDateFormatLong), LongType) AS toDateFormatLong#1681L
    if (assertnotnull(input[0, org.apache.spark.sql.Row, true]).isNullAt) null else validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 4, change_area), LongType) AS change_area#1682L
    if (assertnotnull(input[0, org.apache.spark.sql.Row, true]).isNullAt) null else validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 5, latitude), DoubleType) AS latitude#1683
    if (assertnotnull(input[0, org.apache.spark.sql.Row, true]).isNullAt) null else validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 6, longitude), DoubleType) AS longitude#1684
    if (assertnotnull(input[0, org.apache.spark.sql.Row, true]).isNullAt) null else validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 7, fromDateFormatLong), LongType) AS fromDateFormatLong#1685L
    if (assertnotnull(input[0, org.apache.spark.sql.Row, true]).isNullAt) null else staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 8, iids), StringType), true) AS iids#1686
    if (assertnotnull(input[0, org.apache.spark.sql.Row, true]).isNullAt) null else validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 9, detect_strength), DoubleType) AS detect_strength#1687
    if (assertnotnull(input[0, org.apache.spark.sql.Row, true]).isNullAt) null else staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 10, fromDate), StringType), true) AS fromDate#1688
    if (assertnotnull(input[0, org.apache.spark.sql.Row, true]).isNullAt) null else staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 11, bbox), StringType), true) AS bbox#1689
        at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder.toRow(ExpressionEncoder.scala:290)
        at org.apache.spark.sql.SparkSession$$anonfun$3.apply(SparkSession.scala:573)

Данные возвращаются, если я не включаю bbox.

В браузере neo4j я могу запустить проблемный запрос, и результаты вернутся:

-1  “Detected”  1   20161104    3318    37.5049815  43.4171031  20161023     “filename.val” 9.2 "23OCT16"   [43.4171031, 37.5049815, 43.4171031, 37.5049815]

Это тот вторичный список, мне, возможно, придется вернуть node.bbox.somevalue1 как bbbox1, но я не знаю, каким будет точный синтаксис ....

Я думаю, что это похоже на то, что у меня было ...

Neo4j Spark Connector loadDataFrame выдает ошибку

и решается:

https://github.com/neo4j-contrib/neo4j-spark-connector/issues/40

Просто кажется, что он хочет большего от того, что я возвращаю.


apache-spark scala neo4j geospatial neo4j-spatial
person Codejoy    schedule 20.09.2017    source источник
comment
Можете ли вы поделиться образцом набора данных и точным запросом, который вы выполняете? Спасибо!   -  person Bruno Peres    schedule 26.09.2017
comment
Я отредактировал оригинальный пост, добавив это из первых рук.   -  person Codejoy    schedule 27.09.2017
comment
Я должен добавить, что могу вынуть возврат bbox, и запрос возвращается (хотя я понятия не имею, как получить данные из фрейма данных в виде строк / целых чисел и т. Д., Которые я могу использовать)   -  person Codejoy    schedule 28.09.2017
comment
Мне нужна дополнительная информация?   -  person Codejoy    schedule 29.09.2017
comment
есть ли удача в решении этого вопроса?   -  person Swayam    schedule 16.08.2018

Ответы (2)


arrow_upward
0
arrow_downward

По словам разработчиков, коннектор еще не поддерживает свойства массива (см. https://neo4j-users.slack.com/archives/C0N7LHVS9/p1534429756000100)

Есть два обходных пути:

  1. Используйте UNWIND в запросе, а затем соберите его обратно на Spark
  2. Преобразуйте массив в строку с помощью операции REDUCE

Я предпочитаю второй подход, но не уверен, насколько хорошо он масштабируется с большими данными.

Таким образом, ваш запрос будет выглядеть примерно так:

RETURN REDUCE(s = HEAD(bbox), n IN TAIL(bbox) | s + ', ' + n) AS bbox,

И код, который его обрабатывает:

neo4j.cypher(QUERY).as[String].map(bbox => bbox.split(', ')
person Swayam    schedule 17.08.2018

arrow_upward
-1
arrow_downward

loadDataFrame вам нужно удалить схему фрейма данных с помощью (fieldName и fieldtype)

нравится:

     val rawGraphnode=neo.cypher("MATCH (n:person)where (n.duration <>0) RETURN  n.user as user,n.other as other,n.direction as direction,n.duration as duration,n.timestamp as  timestamp")
            .loadDataFrame(schema = ("user","object"),("other","object"),("direction","string"),("duration","String"),("timestamp","String"))

    rawGraphnode.printSchema()
    rawGraphnode.show(10)
person Muller Haining    schedule 24.02.2018