Spark2.1.0 读取外部txt并以DataFrame输出

以你之姓@ 2022-06-02 05:40 272阅读 0赞

> 需求：使用Spark2.1.0开始Spark的开发，但是网上2.0之后的教程不多，所以自己写一个。  
> ps:官网的教程也有写  
> [http://spark.apache.org/docs/2.1.0/sql-programming-guide.html\#data-sources][http_spark.apache.org_docs_2.1.0_sql-programming-guide.html_data-sources]

外部文件txt,文件内容：全是json串

代码：

import org.apache.spark.api.java.JavaRDD;
    import org.apache.spark.api.java.function.Function;
    import org.apache.spark.sql.Dataset;
    import org.apache.spark.sql.Row;
    import org.apache.spark.sql.SparkSession;
    import org.json.JSONObject;
    
    public class SparkSession2 {
        public static void main(String[] args) {
            SparkSession spark = SparkSession.builder().appName("Spark2").master("local[2]").getOrCreate();
            JavaRDD<String> input = spark.sparkContext().textFile("/data/json/part-r-00000.txt",
                    1).toJavaRDD().map(x -> x.split("____")[1]);
            JavaRDD<Book> books = input.map(new Function<String, Book>() {
                public Book call(String s) throws Exception {
                    JSONObject jsons = new JSONObject(s);
                    Book book = new Book();
    
                    book.setBookid(jsons.get("bookId").toString());
                    book.setConyent(jsons.get("content").toString());
                    book.setContentStartPos(jsons.get("contentStartPos").toString());
                    book.setCoord(jsons.get("coord").toString());
                    book.setId(jsons.get("id").toString());
                    book.setLineColor(jsons.get("lineColor").toString());
                    book.setLineType(jsons.get("lineType").toString());
                    book.setLineWidth(jsons.get("lineWidth").toString());
                    book.setNoteCatalog(jsons.get("noteCatalog").toString());
                    book.setNoteLabels(jsons.get("noteLabels").toString());
                    book.setNoteOrigin(jsons.get("noteOrigin").toString());
                    book.setNotePath(jsons.get("notePath").toString());
                    book.setNotePostil(jsons.get("notePostil").toString());
                    book.setNoteType(jsons.get("noteType").toString());
                    book.setPageAngle(jsons.get("pageAngle").toString());
                    book.setPageHeight(jsons.get("pageHeight").toString());
                    book.setPageIndex(jsons.get("pageIndex").toString());
                    book.setPageWidth(jsons.get("pageWidth").toString());
                    book.setPdfId(jsons.get("pdfId").toString());
                    book.setUpdateTime(jsons.get("updateTime").toString());
                    book.setUserName(jsons.get("userName").toString());
                    book.setSourceType(jsons.get("sourceType").toString());
    
                    return book;
                }
            });
    
            Dataset<Row> bookdf = spark.createDataFrame(books,Book.class);
            bookdf.createOrReplaceTempView("book");
            bookdf.show();
            Dataset<Row> bIddf = spark.sql("select bookid from book");
            bIddf.show();
            spark.close();
        }
    }

> 补：本项目使用idea开发，Spark的standalone模式会产生大量的日志，所以为免输出结果淹没在日志中，设置输出日志为报错日志，其他warning日志不输出。

(maven)在resources下新建log4j.properties

log4j.rootCategory=ERROR, console
    log4j.appender.console=org.apache.log4j.ConsoleAppender
    log4j.appender.console.threshold=INFO
    log4j.appender.console.layout=org.apache.log4j.PatternLayout
    log4j.appender.console.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} [%5p] - %c -%F(%L) -%m%n
    
    log4j.appender.FILE=org.apache.log4j.RollingFileAppender
    log4j.appender.FILE.Append=true
    log4j.appender.FILE.File=logs/log4jtest.log
    log4j.appender.FILE.Threshold=INFO
    log4j.appender.FILE.layout=org.apache.log4j.PatternLayout
    log4j.appender.FILE.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} [%5p] - %c -%F(%L) -%m%n
    log4j.appender.FILE.MaxFileSize=10MB

[http_spark.apache.org_docs_2.1.0_sql-programming-guide.html_data-sources]: http://spark.apache.org/docs/2.1.0/sql-programming-guide.html#data-sources