Как анализировать поля с разделителями, если некоторые (под)поля пусты?

Если ваши данные, как показано ниже в файле

( z287570731_serv80i:7:175 , 5:Re )
( p286274731_serv80i:6:100 , 138 )
( t219420679_serv37i:2:50 , 5 )
( v290380588_serv81i:12:800 , 144:Jo )
( z292902510_serv83i:4:45 , 5:Re )

Затем вы можете использовать

val tid = sc.textFile("path to the input file")
  .map(line => line.split(","))
  .map(array => {
    if (array(1).contains(":")) (array(1).split(":")(1).replace(")", "").trim, array(0).split(":")(1))
    else (null, array(0).split(":")(1))
  })
tid.foreach(println)

который должен дать вам вывод как

(Re,7)
(null,6)
(null,2)
(Jo,12)
(Re,4)

Но если у вас есть данные как

2017-06-18 00:00:00 , 1497769200 , z287570731_serv80i:7:175 , 5:Re
2017-06-18 00:00:00 , 1497769200 , p286274731_serv80i:6:100 , 138
2017-06-18 00:00:00 , 1497769200 , t219420679_serv37i:2:50 , 5
2017-06-18 00:00:00 , 1497769200 , v290380588_serv81i:12:800 , 144:Jo
2017-06-18 00:00:00 , 1497769200 , z292902510_serv83i:4:45 , 5:Re
2017-06-18 00:00:00 , 1497769200 , v205454093_serv75i:5:70 , 50:AK
2017-06-18 00:00:00 , 1497769200 , z287096299_serv80i:19:15000 , 39:Re

Тогда вам нужно сделать

val tid = sc.textFile("path to the input file")
  .map(line => line.split(","))
  .map(array => {
    if (array(3).contains(":")) (array(3).split(":")(1).replace(")", "").trim, array(2).split(":")(1))
    else (null, array(2).split(":")(1))
  })
tid.foreach(println)

И у вас должен быть вывод как

(Re,7)
(null,6)
(null,2)
(Jo,12)
(Re,4)
(AK,5)
(Re,19)

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

Ниже приведена строка, которая вызывает проблему.

{var arr = line.split(","); (arr(3).split(":")(1),arr(2).split(":")(1))});

После того, как вы сделаете line.split(","), вы затем arr(3).split(":")(1), а также arr(2).split(":")(1).

Конечно, слишком много предположений о формате, и его избили пропущенными значениями.

но дает ArrayIndexOutOfBoundsException при выполнении

Это потому, что вы обращаетесь к элементам 3 и 2, но имеете только 2 (!)

scala> sc.textFile("input.csv").
  map { line => line.split(",").toSeq }.
  foreach(println)
WrappedArray(( z287570731_serv80i:7:175i ,  5:Re ))
WrappedArray(( p286274731_serv80i:6:100 ,  138 ))

Проблема почти не связана со Spark. Это обычная проблема Scala, когда данные не там, где вы ожидаете.

scala> val arr = "hello,world".split(",")
arr: Array[String] = Array(hello, world)

Обратите внимание, что то, что выше, — это просто чистый Scala.

Решение 1 — RDD Spark Core

Учитывая следующий набор данных...

2017-06-18 00:00:00 , 1497769200 , z287570731_serv80i:7:175 , 5:Re
2017-06-18 00:00:00 , 1497769200 , p286274731_serv80i:6:100 , 138 
2017-06-18 00:00:00 , 1497769200 , t219420679_serv37i:2:50 , 5
2017-06-18 00:00:00 , 1497769200 , v290380588_serv81i:12:800 , 144:Jo
2017-06-18 00:00:00 , 1497769200 , z292902510_serv83i:4:45 , 5:Re
2017-06-18 00:00:00 , 1497769200 , v205454093_serv75i:5:70 , 50:AK

... Я бы сделал следующее:

val solution = sc.textFile("input.csv").
  map { line => line.split(",") }.
  map { case Array(_, _, third, fourth) => (third, fourth) }.
  map { case (third, fourth) =>
    val Array(_, a @ _*) = fourth.split(":")
    val Array(_, right, _) = third.split(":")
    (a.headOption.orNull, right)
  }
scala> solution.foreach(println)
(Re,7)
(null,6)
(Re,4)
(null,2)
(AK,5)
(Jo,12)

Решение 2 — DataFrames Spark SQL

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

val solution = spark.
  read.
  csv("input.csv").
  select($"_c2" as "third", $"_c3" as "fourth").
  withColumn("a", split($"fourth", ":")).
  withColumn("left", $"a"(1)).
  withColumn("right", split($"third", ":")(1)).
  select("left", "right")
scala> solution.show(false)
+----+-----+
|left|right|
+----+-----+
|Re  |7    |
|null|6    |
|null|2    |
|Jo  |12   |
|Re  |4    |
|AK  |5    |
+----+-----+

ArrayIndexOutOfBounds происходит потому, что элемента не будет, если во втором элементе кортежа нет :. Вы можете проверить, присутствует ли : во втором элементе каждого кортежа. А затем используйте карту, чтобы дать вам промежуточный RDD, на котором вы можете запустить свой текущий запрос.

val rdd = sc.parallelize(Array(
    ( "z287570731_serv80i:7:175" , "5:Re" ),
    ( "p286274731_serv80i:6:100" , "138" ),
    ( "t219420679_serv37i:2:50" , "5" ),
    ( "v290380588_serv81i:12:800" , "144:Jo" ),
    ( "z292902510_serv83i:4:45" , "5:Re" )))


rdd.map { x =>
    val idx = x._2.lastIndexOf(":")
    if(idx == -1) (x._1, x._2+":null")
    else (x._1, x._2)
}

Очевидно, есть лучшие (меньшее количество строк кода) способы сделать то, что вы хотите сделать, но для новичка хорошо размещать каждый шаг в одной команде, чтобы она была легко читаемой и понятной, особенно с scala, где вы можете остановить глобальное потепление с помощью одна строка кода.