java
public class DynamicDemo {
private static SparkConf conf = new SparkConf().setAppName("dynamicdemo").setMaster("local");
private static JavaSparkContext jsc = new JavaSparkContext(conf);
private static SparkSession session = new SparkSession(jsc.sc());
public static void main(String[] args) {
// 创建rdd
JavaRDD<String> rdd = jsc.textFile("./src/main/java/cn/tele/spark_sql/rdd2dataset/students.txt");
// 创建Row的rdd
JavaRDD<Row> rowRdd = rdd.map(new Function<String, Row>() {
private static final long serialVersionUID = 1L;
@Override
public Row call(String v1) throws Exception {
String[] fields = v1.split(",");
return RowFactory.create(Integer.valueOf(fields[0]), fields[1], Integer.valueOf(fields[2]));
}
});
// 创建schema
StructType schema = DataTypes
.createStructType(Arrays.asList(DataTypes.createStructField("id", DataTypes.IntegerType, false),
DataTypes.createStructField("name", DataTypes.StringType, false),
DataTypes.createStructField("age", DataTypes.IntegerType, false)));
// 转换
Dataset<Row> dataset = session.createDataFrame(rowRdd, schema);
dataset.createOrReplaceTempView("students");
Dataset<Row> result = session.sql("select * from students where age<=18");
result.show();
session.stop();
jsc.close();
}
}scala
object DynamicDemo {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("reflectdemo").setMaster("local")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
//创建rdd
val rdd = sc.textFile("./src/main/scala/cn/tele/spark_sql/rdd2dataframe/students.txt", 8)
val rowRdd = rdd.map(lines => {
val arr = lines.split(",");
Row(arr(0).trim().toInt, arr(1), arr(2).trim().toInt)
})
val schema = DataTypes.createStructType(Array(
/* DataTypes.createStructField("id",DataTypes.IntegerType,false),
DataTypes.createStructField("name",DataTypes.StringType,false),
DataTypes.createStructField("age",DataTypes.IntegerType,false)*/
StructField("id", DataTypes.IntegerType, false),
StructField("name", DataTypes.StringType, false),
StructField("age", DataTypes.IntegerType, false)))
//转换
val dataframe = sqlContext.createDataFrame(rowRdd, schema)
dataframe.createOrReplaceTempView("students")
val result = sqlContext.sql("select * from students where age<=18")
result.show()
}
}