import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.List;
import java.util.Random;

public class Test1 {
    public static void main(String[] args)  {
        // (x,y) 元组第一个元素加一个随机的数字前缀

        SparkConf conf = new SparkConf().setAppName("appName").setMaster("local[*]");
        JavaSparkContext sc = new JavaSparkContext(conf);

        List<Tuple2> data = Arrays.asList(new Tuple2(111L, 222L), new Tuple2(100L, 200L));
        JavaRDD<Tuple2> rdd = sc.parallelize(data);


//        JavaPairRDD<String, Long> randomPrefixRdd = rdd.mapToPair
//                (
//                        new PairFunction<Tuple2<Long,Long>, String, Long>()
//                            // input <long,long>
//                            // output typle2(string,long)
//                        {
//                            private static final long serialVersionUID = 1L;
//                            @Override
//                            public Tuple2<String, Long> call(Tuple2<Long, Long> tuple)throws Exception
//                            {
//                                Random random = new Random();
//                                int prefix = random.nextInt(10);
//                                return new Tuple2<String, Long>(prefix + "_" + tuple._1, tuple._2);
//                            }
//                        }
//                );

        JavaPairRDD randomPrefixRdd = rdd.mapToPair( x -> new Tuple2<>(new Random().nextInt(10) + "_" + String.valueOf(x._1), x._2));

        System.out.println(randomPrefixRdd.collect());



    }

}

 

 

 

Tuple2 报错Incompatible equality constrain T1 and String

**解决方式：**可能是spark_core依赖包版本问题，更换一个版本就行了。

19/10/14 16:33:00 ERROR Executor: Exception in task 0.0 in stage 0.0 (TID 0)
java.lang.RuntimeException: Error while encoding: java.lang.RuntimeException: scala.Tuple2 is not a valid external type for schema of string
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]), 0, domain), StringType), true, false) AS domain#0
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, ip), StringType), true, false) AS ip#1
	at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder.toRow(ExpressionEncoder.scala:292)
	at org.apache.spark.sql.SparkSession$$anonfun$4.apply(SparkSession.scala:593)
	at org.apache.spark.sql.SparkSession$$anonfun$4.apply(SparkSession.scala:593)
	at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
	at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
	at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:256)
	at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:247)
	at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:836)
	at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:836)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
	at org.apache.spark.scheduler.Task.run(Task.scala:123)
	at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408)
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.RuntimeException: scala.Tuple2 is not a valid external type for schema of string
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.StaticInvoke_0$(Unknown Source)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply(Unknown Source)
	at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder.toRow(ExpressionEncoder.scala:289)
	... 22 more

解决方案:
字段缺失 与schema不匹配

今天在写spark 提数的时候遇到一个异常，如下

Caused by: java.lang.ClassCastException: org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema cannot be cast to scala.Tuple2
  at $anonfun$1$$anonfun$apply$1.apply(<console>:27)
  at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
  at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
  at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
  at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:35)
  at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
  at scala.collection.AbstractTraversable.map(Traversable.scala:104)
  at $anonfun$1.apply(<console>:27)
  at $anonfun$1.apply(<console>:27)
  at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
  at scala.collection.Iterator$class.foreach(Iterator.scala:893)
  at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
  at org.apache.spark.rdd.RDD$$anonfun$foreach$1$$anonfun$apply$28.apply(RDD.scala:917)
  at org.apache.spark.rdd.RDD$$anonfun$foreach$1$$anonfun$apply$28.apply(RDD.scala:917)
  at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1944)
  at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1944)
  at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
  at org.apache.spark.scheduler.Task.run(Task.scala:99)
  at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:282)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
  at java.lang.Thread.run(Thread.java:745)

df的schema如下：



root
 |-- cityId: long (nullable = true)
 |-- countryId: long (nullable = true)
 |-- outline: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- _1: double (nullable = true)
 |    |    |-- _2: double (nullable = true)
 |-- provinceId: long (nullable = true)
 |-- townId: long (nullable = true)


使用如下方法在提取outline 字段时，报错

val outline = df.select("outline").collect().map(row => row.getAs[Seq[(Double,Double)]]("outline"))

解决方法，我已知两种： 

第一种，在生成数据时，定义case class对象，不存储tuple，就我个人而言，不适用，数据已经生成完毕，再次生成数据需要2天的时间 

第二种方法：



 val lines = json.filter(row => !row.isNullAt(2)).select("outline").rdd.map(r => {
      val row:Seq[(Double,Double)] = r.getAs[Seq[Row]](0).map(x =>{(x.getDouble(0),x.getDouble(1))})
      row
    }).collect()(0)

至于报错的原因，google了一下，我觉得有一种说法可信： 

在row 中的这些列取得时候，要根据类型取，简单的像String，Seq[Double]  这种类型就可以直接取出来，但是像 Seq[(Double,Double)] 这种类型直接取得花就会丢失schema信息，虽然值能取到，但是schema信息丢了，在dataFrame中操作的时候就会抛错