Spark SQL数据源 - 基本操作

news2024/11/22 16:48:50

文章目录

- - 一、案例演示读取Parquet文件
  - 二、在Spark Shell中演示
  - 课堂练习1、将4.1节的student.txt文件转换成student.parquet
  - 课堂练习2、读取student.parquet文件得到学生数据帧，并显示数据帧内容
  - 三、在IntelliJ IDEA里演示

一、案例演示读取Parquet文件

执行命令：cd $SPARK_HOME/examples/src/main/resources，查看Spark的样例数据文件users.parquet
将数据文件users.parquet上传到HDFS的/datasource/input目录

二、在Spark Shell中演示

启动Spark Shell，执行命令：spark-shell --master spark://master:7077
执行命令：val userdf = spark.read.load("hdfs://master:9000/datasource/input/users.parquet")
执行命令：userdf.select("name", "favorite_color").write.save("hdfs://master:9000/datasource/output")，对数据帧指定列进行查询，查询结果依然是数据帧，然后通过write成员的save()方法写入HDFS指定目录
基于数据帧创建临时视图，执行命令：userdf.createTempView("t_user")
执行SQL查询，将结果写入HDFS，执行命令：spark.sql("select name, favorite_color from t_user").write.save("hdfs://master:9000/datasource/output2")
查看HDFS上的输出结果

课堂练习1、将4.1节的student.txt文件转换成student.parquet

解决思路：将student.txt转成studentDF，利用数据帧的save()方法保存到/datasource/output3目录，然后将文件更名复制到/datasource/input目录
得到学生数据帧 - studentDF
将学生数据帧保存为parquet文件
查看生成的parquet文件
复制parquet文件到/datasource/input目录

课堂练习2、读取student.parquet文件得到学生数据帧，并显示数据帧内容

执行命令：val studentDF = spark.read.load("hdfs://master:9000/datasource/input/student.parquet")
执行命令：studentDF.show

三、在IntelliJ IDEA里演示

创建Maven项目
设置项目相关信息
将java目录改成scala目录
在pom.xml文件里添加相关依赖，设置源程序文件夹

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
         http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>net.huawei.sql</groupId>
    <artifactId>SparkSQLDemo</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>2.12.15</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.12</artifactId>
            <version>3.1.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.12</artifactId>
            <version>3.1.3</version>
        </dependency>
    </dependencies>
    <build>
        <sourceDirectory>src/main/scala</sourceDirectory>
    </build>
</project>

在resources目录里添加日志属性文件

log4j.rootLogger=ERROR, stdout, logfile
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/spark.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

在resources目录里添加HFDS配置文件

<?xml version="1.0" encoding="UTF-8"?>
<configuration>
    <property>
        <description>only config in clients</description>
        <name>dfs.client.use.datanode.hostname</name>
        <value>true</value>
    </property>
</configuration>