1

我正在尝试使用 redis 作为 spark sql 的来源，但遇到了如何转换 rdd 的问题。以下是我的代码：

    RDD<Tuple2<String,String>> rdd1 = rc.fromRedisKV("user:*",3,redisConfig);

    JavaRDD<Row> userRDD  = rdd1.toJavaRDD().map(new Function<Tuple2<String,String>, Row>(){
        public Row call(Tuple2<String, String> tuple2) throws Exception {
            System.out.println(tuple2._2);
            return RowFactory.create(tuple2._2().split(","));
        }
    });

    List<StructField> structFields = new ArrayList<StructField>();
    structFields.add(DataTypes.createStructField( "name", DataTypes.StringType, true ));
    structFields.add(DataTypes.createStructField( "sex", DataTypes.StringType, false ));
    structFields.add(DataTypes.createStructField( "age", DataTypes.IntegerType, false ));
    StructType structType = DataTypes.createStructType(structFields);

    Dataset ds = spark.createDataFrame(userRDD, structType);
    ds.createOrReplaceTempView("user");
    ds.printSchema();

    String sql = "select name, sex, age from user ";

    List<Row> list2 = spark.sql(sql).collectAsList();

我得到以下异常：

Caused by: java.lang.ClassCastException: cannot assign instance of scala.collection.immutable.List$SerializationProxy to field org.apache.spark.rdd.RDD.org$apache$spark$rdd$RDD$$dependencies_ of type scala.collection.Seq in instance of org.apache.spark.rdd.MapPartitionsRDD

我不知道下一步该怎么做，请帮忙！

4

1 回答 1

0

我终于找到了原因：我的代码没有问题，但是我需要将我的应用程序的jar上传到Spark服务器。

于 2017-01-16T11:55:50.593 回答