有专门做背景音乐的网站吗,东莞百姓网交友,城乡建设部网官方网站,域名解析怎么设置目录
1.Spark SQL 简介
2.SparkSession
3.Spark SQL 数据的读写
3.1 读写 TXT 文件
3.2 读写 CSV 文件
3.3 读写 JSON 文件
3.4 读写 Parquet 文件
3.5 读写 ORC 文件
3.6 读写MySQL数据库
4.Spark SQL 语法
4.1 SQL 语法
4.2 DSL 语法 1.Spark SQL 简介 Spark SQL…目录
1.Spark SQL 简介
2.SparkSession
3.Spark SQL 数据的读写
3.1 读写 TXT 文件
3.2 读写 CSV 文件
3.3 读写 JSON 文件
3.4 读写 Parquet 文件
3.5 读写 ORC 文件
3.6 读写MySQL数据库
4.Spark SQL 语法
4.1 SQL 语法
4.2 DSL 语法 1.Spark SQL 简介 Spark SQL是Apache Spark的一个模块用于处理结构化数据。它提供了两种编程抽象DataFrame和DataSet并作为分布式SQL查询引擎。
DataFrame DataFrame是一个分布式的数据集合类似于传统数据库中的表带有schema信息。DataFrame是以列式格式进行存储的每一列可以是不同的数据类型如整数、字符串等并且每一列都有一个名称。
DataSet Dataset是在Spark 1.6中新增的一个接口它提供了RDD强类型可以使用强大的lambda函数的优点以及Spark SQL优化执行引擎的优点。数据集可以从JVM对象构建然后使用函数式转换mapflatMapfilter等进行操作。
2.SparkSession SparkSession是Spark 2.0版本引入的新概念它是对SparkContext的升级和扩展。SparkSession提供了更加统一的API在Spark 1.x版本中每个API都需要一个对应的Context例如StreamingContext、SQLContext、HiveContext等包括SQL查询、数据转换、数据读写、注册数据源、管理配置等操作。 SparkSession是Spark 应用程序的新入口点要创建一个基本的SparkSession只需使用SparkSession.builder即可
Java实现代码
package com.yichenkeji.demo.sparkjava;import org.apache.spark.sql.SparkSession;public class SparkSqlDemo {public static void main(String[] args) {SparkSession spark SparkSession.builder().master(local).appName(Spark SQL Demo)
// .config(spark.some.config.option, some-value).getOrCreate();spark.stop();}
}Scala实现代码
package com.yichenkeji.demo.sparkscalaimport org.apache.spark.sql.SparkSessionobject SparkSqlDemo {def main(args: Array[String]): Unit {val spark SparkSession.builder().master(local).appName(Spark SQL Demo)
// .config(spark.some.config.option, some-value).getOrCreate()spark.stop}}3.Spark SQL 数据的读写
3.1 读写 TXT 文件 Spark SQL提供了spark.read().text(file_name)来将文本文件或目录下的文本文件读取到Spark DataFrame中以及dataframe.write().text(path)来将数据写入文本文件。在读取文本文件时每一行默认成为具有字符串“value”列的一行。在写入文本文件时以及dataframe对象中的每一行也只能有一列否则会提示Text data source supports only a single column, and you have 2 columns.错误。
Java代码实现
package com.yichenkeji.demo.sparkjava;import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;public class SparkSqlTxtDemo {public static void main(String[] args) {SparkSession spark SparkSession.builder().master(local).appName(Spark SQL TXT Demo)
// .config(spark.some.config.option, some-value).getOrCreate();//读取txt文件DatasetRow df1 spark.read()
// .option(lineSep, ,)//自定义行分隔符默认是\r, \r\n and \n.text(F:\\24\\spark\\data\\people.txt);df1.show();//读取txt文件的另一种方式DatasetRow df2 spark.read().format(text).load(F:\\24\\spark\\data\\people.txt);df2.show();//写入txt文件df1.write().text(F:\\24\\spark\\data\\people_1.txt);//写入txt文件的另外一种方式df2.write().format(text).save(F:\\24\\spark\\data\\people_2.txt);spark.stop();}
}Scala代码实现
package com.yichenkeji.demo.sparkscalaimport org.apache.spark.sql.SparkSessionobject SparkSqlTxtDemo {def main(args: Array[String]): Unit {val spark SparkSession.builder().master(local).appName(Spark SQL TXT Demo)// .config(spark.some.config.option, some-value).getOrCreate()//读取txt文件val df1 spark.read.text(F:\\24\\spark\\data\\people.txt)df1.show//读取txt文件的另一种方式val df2 spark.read.format(text).load(F:\\24\\spark\\data\\people.txt)df2.show//写入txt文件df1.write.text(F:\\24\\spark\\data\\people_1.txt)//写入txt文件的另一种方式df2.write.format(text).save(F:\\24\\spark\\data\\people_2.txt)spark.stop}}3.2 读写 CSV 文件 Spark SQL提供了spark.read().csv(file_name)将CSV格式的文件或文件夹读取到Spark DataFrame中以及dataframe.write().csv(path)将数据写入CSV文件。
Java代码实现
package com.yichenkeji.demo.sparkjava;import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;public class SparkSqlCsvDemo {public static void main(String[] args) {SparkSession spark SparkSession.builder().master(local).appName(Spark SQL CSV Demo)
// .config(spark.some.config.option, some-value).getOrCreate();//读取CSV文件CSV带有表头DatasetRow df1 spark.read().option(header, true)//设置是否有表头.option(delimiter, ;) //设置分隔符默认是逗号.csv(F:\\24\\spark\\data\\people.csv);df1.show();//读取CSV文件CSV带有表头//csv文件的另外一种读取方式DatasetRow df2 spark.read().format(csv).option(header, true)//设置是否有表头.option(delimiter, ;) //设置分隔符默认是逗号.load(F:\\24\\spark\\data\\people.csv);df2.show();//写入CSV文件df1.write().csv(F:\\24\\spark\\data\\people_1.csv);//写入CSV文件的另外一种方式df2.write().format(csv).save(F:\\24\\spark\\data\\people_2.csv);spark.stop();}
}Scala代码实现
package com.yichenkeji.demo.sparkscalaimport org.apache.spark.sql.SparkSessionobject SparkSqlCsvDemo {def main(args: Array[String]): Unit {val spark SparkSession.builder().master(local).appName(Spark SQL CSV Demo)// .config(spark.some.config.option, some-value).getOrCreate()//读取CSV文件val df1 spark.read.option(header, true)//是否有表头.option(delimiter, ;) //分隔符.csv(F:\\24\\spark\\data\\people.csv)df1.show//读取CSV文件的另一种方式val df2 spark.read.format(csv).option(header, true) //是否有表头.option(delimiter, ;) //分隔符.load(F:\\24\\spark\\data\\people.csv)df2.show//写入csv文件df1.write.csv(F:\\24\\spark\\data\\people_1.csv)//写入csv文件的另一种方式df2.write.format(csv).save(F:\\24\\spark\\data\\people_2.csv)spark.stop}}3.3 读写 JSON 文件 Spark SQL提供了spark.read().json(file_name)将json格式的文件或文件夹读取到Spark DataFrame中以及dataframe.write().json(path)将数据写入json文件。
Java代码实现
package com.yichenkeji.demo.sparkjava;import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;public class SparkSqlJsonDemo {public static void main(String[] args) {SparkSession spark SparkSession.builder().master(local).appName(Spark SQL JSON Demo)
// .config(spark.some.config.option, some-value).getOrCreate();//读取JSON文件DatasetRow df1 spark.read().json(F:\\24\\spark\\data\\people.json);df1.show();//读取JSON文件//JSON文件的另外一种读取方式DatasetRow df2 spark.read().format(json).load(F:\\24\\spark\\data\\people.json);df2.show();//写入JSON文件df1.write().json(F:\\24\\spark\\data\\people_1.json);//写入JSON文件的另外一种方式df2.write().format(csv).save(F:\\24\\spark\\data\\people_2.json);spark.stop();}
}Scala代码实现
package com.yichenkeji.demo.sparkscalaimport org.apache.spark.sql.SparkSessionobject SparkSqlJsonDemo {def main(args: Array[String]): Unit {val spark SparkSession.builder().master(local).appName(Spark SQL JSON Demo)// .config(spark.some.config.option, some-value).getOrCreate()//读取JSON文件val df1 spark.read.json(F:\\24\\spark\\data\\people.json)df1.show//读取JSON文件的另一种方式val df2 spark.read.format(json).load(F:\\24\\spark\\data\\people.json)df2.show//写入JSON文件df1.write.json(F:\\24\\spark\\data\\people_1.json)//写入JSON件的另一种方式df2.write.format(json).save(F:\\24\\spark\\data\\people_2.json)spark.stop}}3.4 读写 Parquet 文件 Parquet 是一种通用的列式存储格式Parquet已广泛应用于Hadoop生态圈(MapReduce、Spark、Hive、Impala)是离线数仓的主要数据格式。Spark SQL提供了spark.read().parquet(file_name)将parquet格式的文件或文件夹读取到Spark DataFrame中以及dataframe.write().parquet(path)将数据写入parquet文件。 Java代码实现
package com.yichenkeji.demo.sparkjava;import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;public class SparkSqlParquetDemo {public static void main(String[] args) {SparkSession spark SparkSession.builder().master(local).appName(Spark SQL Parquet Demo)
// .config(spark.some.config.option, some-value).getOrCreate();//读取Parquet文件DatasetRow df1 spark.read().parquet(F:\\24\\spark\\data\\users.parquet);df1.show();//Parquet件的另外一种读取方式DatasetRow df2 spark.read().format(parquet).load(F:\\24\\spark\\data\\users.parquet);df2.show();//写入Parquet文件df1.write().parquet(F:\\24\\spark\\data\\users_1.parquet);//写入Parquet文件的另外一种方式df2.write().format(parquet).save(F:\\24\\spark\\data\\users_2.parquet);spark.stop();}
}Scala代码实现
package com.yichenkeji.demo.sparkscalaimport org.apache.spark.sql.SparkSessionobject SparkSqlParquetDemo {def main(args: Array[String]): Unit {val spark SparkSession.builder().master(local).appName(Spark SQL Parquet Demo)// .config(spark.some.config.option, some-value).getOrCreate()//读取Parquet文件val df1 spark.read.parquet(F:\\24\\spark\\data\\users.parquet)df1.show//读取Parquet文件的另一种方式val df2 spark.read.format(parquet).load(F:\\24\\spark\\data\\users.parquet)df2.show//写入Parquet文件df1.write.parquet(F:\\24\\spark\\data\\users_1.parquet)//写入Parquet件的另一种方式df2.write.format(parquet).save(F:\\24\\spark\\data\\users_2.parquet)spark.stop}}3.5 读写 ORC 文件 ORC是一种列式存储结构ORC最早产生于Apache Hive用于减小Hadoop文件的存储空间和加速Hive查询。和Parquet一样ORC也广泛应用于Hadoop生态圈。Spark SQL提供了spark.read().orc(file_name)将orc格式的文件或文件夹读取到Spark DataFrame中以及dataframe.write().orc(path)将数据写入orc文件。 Java代码实现
package com.yichenkeji.demo.sparkjava;import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;public class SparkSqlOrcDemo {public static void main(String[] args) {SparkSession spark SparkSession.builder().master(local).appName(Spark SQL ORC Demo)
// .config(spark.some.config.option, some-value).getOrCreate();//读取ORC文件DatasetRow df1 spark.read().orc(F:\\24\\spark\\data\\users.orc);df1.show();//ORC文件的另外一种读取方式DatasetRow df2 spark.read().format(orc).load(F:\\24\\spark\\data\\users.orc);df2.show();//写入ORC文件df1.write().orc(F:\\24\\spark\\data\\users_1.orc);//写入ORC文件的另外一种方式df2.write().format(orc).save(F:\\24\\spark\\data\\users_2.orc);spark.stop();}
}Scala代码实现
package com.yichenkeji.demo.sparkscalaimport org.apache.spark.sql.SparkSessionobject SparkSqlOrcDemo {def main(args: Array[String]): Unit {val spark SparkSession.builder().master(local).appName(Spark SQL ORC Demo)// .config(spark.some.config.option, some-value).getOrCreate()//读取ORC文件val df1 spark.read.orc(F:\\24\\spark\\data\\users.orc)df1.show//读取ORC文件的另一种方式val df2 spark.read.format(orc).load(F:\\24\\spark\\data\\users.orc)df2.show//写入ORC文件df1.write.orc(F:\\24\\spark\\data\\users_1.orc)//写入ORC件的另一种方式df2.write.format(orc).save(F:\\24\\spark\\data\\users_2.orc)spark.stop}}3.6 读写MySQL数据库 Spark SQL 可以通过 JDBC 从关系型数据库中读取数据的方式创建 DataFrame通过对DataFrame 一系列的计算后还可以将数据再写回关系型数据库中。
Java代码实现
package com.yichenkeji.demo.sparkjava;import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;import java.util.Properties;public class SparkSqlMySQLDemo {public static void main(String[] args) {SparkSession spark SparkSession.builder().master(local).appName(Spark SQL MySQL Demo)
// .config(spark.some.config.option, some-value).getOrCreate();String url jdbc:mysql://localhost:3307/demo?useSSLfalse;String user root;String password wsx-123;String driver com.mysql.cj.jdbc.Driver;String table demo.dim_area;Properties properties new Properties();properties.setProperty(user,user);properties.setProperty(password,password);properties.setProperty(driver,driver);properties.setProperty(url,url);properties.setProperty(useUnicode,true);properties.setProperty(characterEncoding,utf-8);//读取MySQL文件DatasetRow df1 spark.read().jdbc(url,table,properties);df1.show(2);//MySQL的另外一种读取方式DatasetRow df2 spark.read().format(jdbc).option(url, url).option(dbtable, table).option(user, user).option(password, password).load();df2.show(3);// //写入MySQL文件df1.write().mode(SaveMode.Overwrite).jdbc(url,dim_area_1,properties);
// //写入MySQL的另外一种方式df2.write().mode(SaveMode.Overwrite).format(jdbc).option(url, url).option(dbtable, dim_area_2).option(user, user).option(password, password).save();spark.stop();}
}Scala代码实现
package com.yichenkeji.demo.sparkscalaimport org.apache.spark.sql.{Dataset, Row, SaveMode, SparkSession}import java.util.Propertiesobject SparkSqlMySQLDemo {def main(args: Array[String]): Unit {val spark SparkSession.builder().master(local).appName(Spark SQL MySQL Demo)// .config(spark.some.config.option, some-value).getOrCreate()val url jdbc:mysql://localhost:3307/demo?useSSLfalseval user rootval password wsx-123val driver com.mysql.cj.jdbc.Driverval table demo.dim_areaval properties new Propertiesproperties.setProperty(user, user)properties.setProperty(password, password)properties.setProperty(driver, driver)properties.setProperty(url, url)properties.setProperty(useUnicode, true)properties.setProperty(characterEncoding, utf-8)//读取MySQL文件val df1 spark.read.jdbc(url, table, properties)df1.show(2)//MySQL的另外一种读取方式val df2 spark.read.format(jdbc).option(url, url).option(dbtable, table).option(user, user).option(password, password).loaddf2.show(3)//写入MySQL文件df1.write.mode(SaveMode.Overwrite).jdbc(url, dim_area_1, properties)//写入MySQL的另外一种方式df2.write.mode(SaveMode.Overwrite).format(jdbc).option(url, url).option(dbtable, dim_area_2).option(user, user).option(password, password).savespark.stop}}注意通过jdbc方式写入mysql数据库是dataframe中的字段在保存的表中必须存在否则报错。 4.Spark SQL 语法
4.1 SQL 语法 SparkSession中的sql函数允许应用程序编程地运行SQL查询并将结果作为DataFrame返回。这种方式的查询必须要有临时视图或者全局视图来辅助。
Java代码实现
package com.yichenkeji.demo.sparkjava;import org.apache.spark.sql.AnalysisException;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;public class SparkSqlDemo {public static void main(String[] args) throws AnalysisException {SparkSession spark SparkSession.builder().master(local).appName(Spark SQL Demo)
// .config(spark.some.config.option, some-value).getOrCreate();//读取CSV文件CSV带有表头DatasetRow df1 spark.read().option(header, true)//设置是否有表头.option(delimiter, ;) //设置分隔符默认是逗号.csv(F:\\24\\spark\\data\\people.csv);df1.show();//创建临时视图df1.createOrReplaceTempView(people);//创建全局临时视图df1.createGlobalTempView(people);DatasetRow sqlDF spark.sql(SELECT * FROM people);sqlDF.show();//-----------------//| name|age| job|//-----------------//|Jorge| 30|Developer|//| Bob| 32|Developer|//-----------------//全局临时视图保存在global_temp数据库中使用全局临时视图时需要全路径访问如global_temp.peopleDatasetRow sqlDF2 spark.sql(SELECT name,age,job FROM global_temp.people WHERE age 30);sqlDF2.show();//-----------------//| name|age| job|//-----------------//| Bob| 32|Developer|//-----------------spark.stop();}
}Scala代码实现
package com.yichenkeji.demo.sparkscalaimport org.apache.spark.sql.SparkSessionobject SparkSqlDemo {def main(args: Array[String]): Unit {val spark SparkSession.builder().master(local).appName(Spark SQL Demo)
// .config(spark.some.config.option, some-value).getOrCreate()//读取CSV文件val df1 spark.read.option(header, true) //是否有表头.option(delimiter, ;) //分隔符.csv(F:\\24\\spark\\data\\people.csv)df1.show// 创建临时视图df1.createOrReplaceTempView(people)// 创建全局临时视图df1.createGlobalTempView(people)//使用临时视图spark.sql(SELECT * FROM people).show()//-----------------//| name|age| job|//-----------------//|Jorge| 30|Developer|//| Bob| 32|Developer|//-----------------// 全局临时视图保存在global_temp数据库中使用全局临时视图时需要全路径访问如global_temp.peoplespark.sql(SELECT name,age,job FROM global_temp.people where age 30).show()//-----------------//| name|age| job|//-----------------//| Bob| 32|Developer|//-----------------spark.stop}}4.2 DSL 语法 DataFrame DataFrame为Scala、Java、Python和R提供了针对结构化数据操作的领域特定语言。使用 DSL 语法不必创建临时视图。
Java代码实现
package com.yichenkeji.demo.sparkjava;import org.apache.spark.sql.*;import static org.apache.spark.sql.functions.col;
public class SparkSqlDSLDemo {public static void main(String[] args) throws AnalysisException {SparkSession spark SparkSession.builder().master(local).appName(Spark SQL DSL)
// .config(spark.some.config.option, some-value).getOrCreate();//读取CSV文件CSV带有表头DatasetRow df1 spark.read().option(header, true)//设置是否有表头.option(delimiter, ;) //设置分隔符默认是逗号.csv(F:\\24\\spark\\data\\people.csv);df1.show();// 打印字段信息df1.printSchema();
// root
// |-- name: string (nullable true)
// |-- age: string (nullable true)
// |-- job: string (nullable true)//查询指定字段df1.select(name,age,job).show();//-----------------//| name|age| job|//-----------------//|Jorge| 30|Developer|//| Bob| 32|Developer|//-----------------//查询指定字段并设置别名对查询的字段进行运算df1.select(col(name).alias(name),col(age).plus(1).alias(age),col(job)).show();//-----------------//| name|age| job|//-----------------//|Jorge| 31|Developer|//| Bob| 33|Developer|//-----------------//过滤数据df1.filter(col(age).gt(30)).show();
// df1.where(col(age).gt(30)).show();//-----------------//| name|age| job|//-----------------//| Bob| 32|Developer|//-----------------//分组聚合df1.groupBy(job).count().show();//--------------//| job|count|//--------------//|Developer| 2|//--------------//复杂聚合df1.groupBy(col(job).alias(post)).agg(functions.count(col(name)).as(nums),functions.sum(col(age)).as(total_age),functions.avg(col(age)).as(aver_age)).show();//------------------------------//| post|nums|total_age|aver_age|//------------------------------//|Developer| 2| 62.0| 31.0|//------------------------------spark.stop();}
}Scala代码实现
package com.yichenkeji.demo.sparkscalaimport org.apache.spark.sql.{SparkSession, functions}
import org.apache.spark.sql.functions.colobject SparkSqlDSLDemo {def main(args: Array[String]): Unit {val spark SparkSession.builder().master(local).appName(Spark SQL DSL)
// .config(spark.some.config.option, some-value).getOrCreate()//读取CSV文件val df1 spark.read.option(header, true) //是否有表头.option(delimiter, ;) //分隔符.csv(F:\\24\\spark\\data\\people.csv)df1.show//打印字段信息df1.printSchema// root// |-- name: string (nullable true)// |-- age: string (nullable true)// |-- job: string (nullable true)//查询指定字段df1.select(name, age, job).show//-----------------//| name|age| job|//-----------------//|Jorge| 30|Developer|//| Bob| 32|Developer|//-----------------//查询指定字段并设置别名对查询的字段进行运算import spark.implicits._df1.select($name as name, ($age 1) as age, $job as job).show//-----------------//| name|age| job|//-----------------//|Jorge| 31|Developer|//| Bob| 33|Developer|//-----------------//过滤数据df1.filter($age 30).show
// df1.where($age 30).show//-----------------//| name|age| job|//-----------------//| Bob| 32|Developer|//-----------------//分组聚合df1.groupBy(job).count.show//--------------//| job|count|//--------------//|Developer| 2|//--------------//复杂聚合df1.groupBy(col(job).alias(post)).agg(functions.count(col(name)).as(nums), functions.sum(col(age)).as(total_age), functions.avg(col(age)).as(aver_age)).show()//------------------------------//| post|nums|total_age|aver_age|//------------------------------//|Developer| 2| 62.0| 31.0|//------------------------------spark.stop}}
