2016-08-19 3 views
-4

Я пытаюсь напечатать содержимое моего РДУ RDD[(String,List[(String,String)])]:Невозможно напечатать содержимое РДУ

val sc = new SparkContext(conf) 
val splitted = rdd.map(line => line.split(",")) 
val processed = splitted.map(x=>(x(1),List((x(0),x(2),x(3),x(4))))) 
val grouped = processed.reduceByKey((x,y) => (x ++ y)) 
System.out.println(grouped) 

Однако вместо того, чтобы содержимое я вижу:

ShuffledRDD[4] at reduceByKey at Consumer.scala:88

UPDATE:

Содержание TXT-файла:

100001082016,230,111,1,1 
100001082016,121,111,1,1 
100001082016,110,111,1,1 

UPDATE 2 (весь код):

class Consumer() 
{ 

def run() = { 
    val conf = new SparkConf() 
           .setAppName("TEST") 
           .setMaster("local[*]") 
    val sc = new SparkContext(conf) 
    val rdd = sc.textFile("file:///usr/test/myfile.txt") 
    val splitted = rdd.map(line => line.split(",")) 
    val processed = splitted.map(x=>(x(1),List((x(0),x(2),x(3),x(4))))) 
    val grouped = processed.reduceByKey((x,y) => (x ++ y)) 
    System.out.println(grouped) 
} 

} 
+0

Что говорит остальная часть stacktrace? –

+0

И в Scala вы будете делать 'println (grouped.collect())'. Нет необходимости в System.out –

+0

@ cricket_007: В этом случае я получаю '[Lscala.Tuple2; @ 5377414a'. Остальная часть стека является стандартным выходом Spark, например. '6/08/19 13:49:39 INFO DAGScheduler: Job 0 done: collect at Consumer.scala: 89, взял 0.519500 s' и т. Д. – HackerDuck

ответ

3

Там нет вопроса здесь:

scala> val rdd = sc.parallelize(Seq("100001082016,230,111,1,1","100001082016,121,111,1,1","100001082016,110,111,1,1")) 
// rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[0] at parallelize at <console>:27 

scala> val splitted = rdd.map(line => line.split(",")) 
// splitted: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[1] at map at <console>:29 

scala> val processed = splitted.map(x=>(x(1),List((x(0),x(2),x(3),x(4))))) 
// processed: org.apache.spark.rdd.RDD[(String, List[(String, String, String, String)])] = MapPartitionsRDD[2] at map at <console>:31 

scala> val grouped = processed.reduceByKey((x,y) => (x ++ y)) 
// grouped: org.apache.spark.rdd.RDD[(String, List[(String, String, String, String)])] = ShuffledRDD[3] at reduceByKey at <console>:33 

scala> grouped.collect().foreach(println) 
// (121,List((100001082016,111,1,1))) 
// (110,List((100001082016,111,1,1))) 
// (230,List((100001082016,111,1,1))) 

Следующая неправильно. Он работает, как ожидают, но вы должны понимать язык правильно, чтобы знать, что ожидать:

scala> System.out.println(grouped) 
// ShuffledRDD[3] at reduceByKey at <console>:33 

EDIT: Просто чтобы быть ясно, если вы хотите напечатать коллекцию, вы должны будете использовать mkString доступен для коллекции, которую вам нужно будет преобразовать в формат, который вы хотите.

+0

'grouped.collect()' работает в консоли, потому что он рассматривает печать массива специально. 'println (grouped.collect())' _doesn't_ работает по желанию: просто попробуйте 'println (Array (1,2))'. –

+0

@AlexeyRomanov Я ничего не говорил о 'println (grouped.collect)'. Still println (grouped) будет использовать метод 'toString'' ShuffledRDD'. Я думаю, что мы оба согласны с этим, поэтому работает так, как и ожидалось. – eliasah

+0

@ AlexeyRomanov grouped.collect здесь был просто примером, может быть, я должен удалить его, чтобы избежать путаницы. – eliasah

Смежные вопросы