Spark向HDFS中存储数据

时间：2024-01-19 18:52:56

程序如下：

import org.apache.spark.sql.Row;

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaRDD;

import org.apache.spark.api.java.JavaSparkContext;

import org.apache.spark.api.java.function.VoidFunction;

import org.apache.spark.sql.DataFrame;

import org.apache.spark.sql.SQLContext;

public class QueryAndStoreHDFSData {

    static SparkConf sparkConf = new SparkConf().setAppName("HDFSQuery").setMaster("local[2]");

    static JavaSparkContext sc = new JavaSparkContext(sparkConf);

    static SQLContext sqlContext = new SQLContext(sc);

    public static void main(String[] args){

//        JavaRDD<String> poi = sc.textFile("hdfs://node2:9000/user/flume/events/2015-11-27-21/events-.1448629506841");

        DataFrame df = sqlContext.read().json("hdfs://node2:9000/user/flume/events/2015-11-26-21/events-.1448543965316");

        // 打印模式

        df.printSchema();

        // 将数据框架注册成一个表

        df.registerTempTable("poi");

        // 使用sql语句从表中读取数据

        DataFrame poi = sqlContext.sql("SELECT * FROM poi WHERE cid=57425749418");

        JavaRDD<Row> row = poi.javaRDD();

        //将RDD中的数据存入HDFS（也可以指定其他目录和格式）

        row.saveAsTextFile("hdfs://node2:9000/user/poi.txt");

        row.foreach(new VoidFunction<Row>(){

            @Override

            public void call(Row r) throws Exception {

                System.out.println(r.mkString());

            }

        });

    }

}

相关文章