Spark数据源

刺骨的言语ヽ痛彻心扉 2022-05-14 05:45 184阅读 0赞

教程：

[http://spark.apache.org/docs/latest/sql-programming-guide.html\#hive-metastore-parquet-table-conversion][http_spark.apache.org_docs_latest_sql-programming-guide.html_hive-metastore-parquet-table-conversion]

### 多种类型数据源加载和保存 ###

// $example on:generic_load_save_functions$
        Dataset<Row> usersDF = spark.read().load("examples/src/main/resources/users.parquet");
        usersDF.select("name", "favorite_color").write().save("namesAndFavColors.parquet");
        // $example off:generic_load_save_functions$
        // $example on:manual_load_options$
        Dataset<Row> peopleDF = spark.read().format("json").load("examples/src/main/resources/people.json");
        peopleDF.select("name", "age").write().format("parquet").save("namesAndAges.parquet");
        // $example off:manual_load_options$
        // $example on:manual_load_options_csv$
        Dataset<Row> peopleDFCsv = spark.read().format("csv")
                .option("sep", ";")
                .option("inferSchema", "true")
                .option("header", "true")
                .load("examples/src/main/resources/people.csv");
        // $example off:manual_load_options_csv$
        // $example on:direct_sql$
        Dataset<Row> sqlDF =
                spark.sql("SELECT * FROM parquet.`examples/src/main/resources/users.parquet`");
        // $example off:direct_sql$
        // $example on:write_sorting_and_bucketing$
        peopleDF.write().bucketBy(42, "name").sortBy("age").saveAsTable("people_bucketed");
        // $example off:write_sorting_and_bucketing$
        // $example on:write_partitioning$
        usersDF.write()
                .partitionBy("favorite_color")
                .format("parquet")
                .save("namesPartByColor.parquet");
        // $example off:write_partitioning$
        // $example on:write_partition_and_bucket$
        peopleDF.write()
                .partitionBy("favorite_color")
                .bucketBy(42, "name")
                .saveAsTable("people_partitioned_bucketed");
        // $example off:write_partition_and_bucket$
    
        spark.sql("DROP TABLE IF EXISTS people_bucketed");
        spark.sql("DROP TABLE IF EXISTS people_partitioned_bucketed");

### 连接jdbc ###

Dataset<Row> jdbcDF = spark.read()
                .format("jdbc")
                .option("url", "jdbc:postgresql:dbserver")
                .option("dbtable", "schema.tablename")
                .option("user", "username")
                .option("password", "password")
                .load();
    
        Properties connectionProperties = new Properties();
        connectionProperties.put("user", "username");
        connectionProperties.put("password", "password");
        Dataset<Row> jdbcDF2 = spark.read()
                .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties);
    
        // Saving data to a JDBC source
        jdbcDF.write()
                .format("jdbc")
                .option("url", "jdbc:postgresql:dbserver")
                .option("dbtable", "schema.tablename")
                .option("user", "username")
                .option("password", "password")
                .save();
    
        jdbcDF2.write()
                .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties);
    
        // Specifying create table column data types on write
        jdbcDF.write()
                .option("createTableColumnTypes", "name CHAR(64), comments VARCHAR(1024)")
                .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties);

### spark连接hive ###

与hive的兼容性

[http://spark.apache.org/docs/latest/sql-programming-guide.html\#compatibility-with-apache-hive][http_spark.apache.org_docs_latest_sql-programming-guide.html_compatibility-with-apache-hive]

package com.kexin.spak;
    
    
    import java.io.File;
    import java.io.Serializable;
    import java.util.ArrayList;
    import java.util.List;
    
    import org.apache.spark.api.java.function.MapFunction;
    import org.apache.spark.sql.Dataset;
    import org.apache.spark.sql.Encoders;
    import org.apache.spark.sql.Row;
    import org.apache.spark.sql.SparkSession;
    
    /**
     * @Author KeXin
     * @Date 2018/8/28 下午3:23
     **/
    
    public  class SparkHive implements Serializable {
        
    
        public static void main(String[] args) {
            // warehouseLocation points to the default location for managed databases and tables
            String warehouseLocation = new File("spark-warehouse").getAbsolutePath();
            SparkSession spark = SparkSession
                    .builder()
                    .appName("Java Spark Hive Example")
                    .config("spark.sql.warehouse.dir", warehouseLocation)
                    .master("local")
                    .enableHiveSupport()
                    .getOrCreate();
    
            spark.sql("CREATE TABLE IF NOT EXISTS testSpark (key INT, value STRING) USING hive");
            spark.sql("LOAD DATA LOCAL INPATH '/Users/kexin/work/projects/TestSpark/src/main/resources/kv1.txt' INTO TABLE testSpark");
    
            // Queries are expressed in HiveQL
            spark.sql("SELECT * FROM testSpark").show();
    
            // Aggregation queries are also supported.
            spark.sql("SELECT COUNT(*) FROM testSpark").show();
    
            // The results of SQL queries are themselves DataFrames and support all normal functions.
            Dataset<Row> sqlDF = spark.sql("SELECT key, value FROM testSpark WHERE key < 10 ORDER BY key");
    
            // The items in DataFrames are of type Row, which lets you to access each column by ordinal.
            Dataset<String> stringsDS = sqlDF.map(
                    (MapFunction<Row, String>) row -> "Key: " + row.get(0) + ", Value: " + row.get(1),
                    Encoders.STRING());
            stringsDS.show();
            // You can also use DataFrames to create temporary views within a SparkSession.
            List<Record> records = new ArrayList<>();
            for (int key = 1; key < 100; key++) {
                Record record = new Record();
                record.setKey(key);
                record.setValue("val_" + key);
                records.add(record);
            }
            Dataset<Row> recordsDF = spark.createDataFrame(records, Record.class);
            recordsDF.createOrReplaceTempView("records");
    
            // Queries can then join DataFrames data with data stored in Hive.
            spark.sql("SELECT * FROM records r JOIN testSpark s ON r.key = s.key").show();
            spark.stop();
        }
    
        public static class Record implements Serializable {
        
            private int key;
            private String value;
    
            public int getKey() {
                return key;
            }
    
            public void setKey(int key) {
                this.key = key;
            }
    
            public String getValue() {
                return value;
            }
    
            public void setValue(String value) {
                this.value = value;
            }
        }
    }

[http_spark.apache.org_docs_latest_sql-programming-guide.html_hive-metastore-parquet-table-conversion]: http://spark.apache.org/docs/latest/sql-programming-guide.html#hive-metastore-parquet-table-conversion
[http_spark.apache.org_docs_latest_sql-programming-guide.html_compatibility-with-apache-hive]: http://spark.apache.org/docs/latest/sql-programming-guide.html#compatibility-with-apache-hive