apache-spark - 如何将redis转成spark数据集或dataframe？

Question

我正在尝试使用 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

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

Answer 1

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