spark读取txt文件

### 使用 Spark 读取 TXT 文件并将数据写入 MySQL 数据库 在 Spark 中，可以通过 `SparkSession` 对象使用 SQL 操作来读取 TXT 文件，并将数据写入 MySQL 数据库。以下是实现此功能的完整步骤和代码示例。 #### 环境准备 1. **安装 MySQL**：确保 MySQL 数据库已安装并运行。 2. **创建 MySQL 数据库和表**： ```sql CREATE DATABASE sparkdb; USE sparkdb; CREATE TABLE users ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(50), age INT, country VARCHAR(50) ); ``` 3. **下载 MySQL JDBC 驱动**：从 MySQL 官方网站下载 MySQL JDBC 驱动（如 `mysql-connector-java-x.x.xx.jar`），并将该 JAR 文件放置在 Spark 的 `jars` 目录下（例如 `spark-3.3.0/jars/`）。 #### 示例代码 以下是一个完整的代码示例，展示如何使用 PySpark 读取 TXT 文件并将数据写入 MySQL 数据库[^2]。 ```python from pyspark.sql import SparkSession # 创建 SparkSession 对象 spark = SparkSession.builder \ .appName("TXT to Database") \ .getOrCreate() # 读取 TXT 文件 df = spark.read \ .format("text") \ .load("/path/to/file/data.txt") # 假设 TXT 文件中的每一行是逗号分隔的字符串，可以对其进行解析 df = df.selectExpr("split(value, ',') as columns") \ .selectExpr("columns[0] as name", "columns[1] as age", "columns[2] as country") # 打印列名及数据类型 df.printSchema() # 显示前几行数据 df.show() # 设置数据库连接信息 jdbc_url = "jdbc:mysql://hostname:port/database_name" db_properties = { "user": "username", "password": "password", "driver": "com.mysql.jdbc.Driver" } # 将数据写入 MySQL 数据库 df.write \ .format("jdbc") \ .option("url", jdbc_url) \ .option("dbtable", "users") \ .mode("append") \ .options(**db_properties) \ .save() ``` #### 代码说明 1. **读取 TXT 文件**：通过 `spark.read.format("text").load()` 方法读取 TXT 文件，每行作为一个字符串。 2. **解析数据**：假设 TXT 文件中的每一行是逗号分隔的字符串，可以使用 `split` 函数将其解析为多个字段。 3. **写入 MySQL 数据库**：通过 `df.write.format("jdbc")` 方法将数据写入 MySQL 数据库，需指定 JDBC URL 和数据库连接属性。 #### 注意事项 - 如果 TXT 文件的格式不是逗号分隔的字符串，则需要根据实际格式调整解析逻辑。 - 确保 MySQL JDBC 驱动已正确配置到 Spark 的 `jars` 目录下，否则可能会导致驱动类找不到的错误。

### 使用 Apache Spark 读取和写入 CSV 文件 Apache Spark 提供了多种方式来读取和写入 CSV 文件。在不同版本的 Spark 中，操作方法略有不同。以下将详细介绍如何使用 Spark 2.0+ 版本进行 CSV 文件的读写操作。 #### 1. 读取 CSV 文件 在 Spark 2.0 及以上版本中，CSV 文件的读取可以通过 `SparkSession` 来完成，无需额外添加 Maven 依赖。核心代码如下： ```scala val spark = SparkSession .builder() .master("local[2]") .appName("app") .getOrCreate() // 读取 CSV 文件 val srcDF = spark.read .format("csv") .option("header", "true") // 指定第一行为表头 .option("multiLine", "true") // 允许字段中包含换行符 .load("file:///C:\\1.csv") ``` - `.option("header", "true")` 表示文件的第一行是列名。 - `.option("multiLine", "true")` 允许字段中包含换行符[^2]。 如果需要手动指定列名，可以使用 `.toDF()` 方法： ```scala val data = spark.read .option("inferSchema", "true") // 自动推断数据类型 .option("header", "false") // 不处理表头 .csv("file:///home/spark/file/project/visit.txt") .toDF("mac", "phone_brand", "enter_time", "first_time", "last_time", "region", "screen", "stay_time") ``` - `.option("inferSchema", "true")` 表示自动推断数据类型[^3]。 #### 2. 写入 CSV 文件 写入 CSV 文件时，可以通过 `DataFrame` 的 `write` 方法实现。以下是一个完整的示例： ```scala val localpath = "D:\\input\\word.csv" val outpath = "D:\\output\\word2" val conf = new SparkConf() conf.setAppName("SparkReadFile") conf.setMaster("local") val sparkContext = new SparkContext(conf) val sqlContext = new SQLContext(sparkContext) // 读取 CSV 文件 val data: DataFrame = sqlContext.read.format("com.databricks.spark.csv") .option("header", "false") // 不处理表头 .option("inferSchema", "true") // 自动推断数据类型 .load(localpath) // 写入 CSV 文件 data.repartition(1).write.format("com.databricks.spark.csv") .option("header", "false") // 不输出表头 .option("delimiter", ",") // 设置分隔符为逗号 .save(outpath) ``` - `.repartition(1)` 用于将输出文件的数量设置为 1，避免生成多个小文件。 - `.option("header", "false")` 表示不输出表头。 - `.option("delimiter", ",")` 设置字段之间的分隔符为逗号[^1]。 #### 3. 注意事项 - 在 Spark 2.0 之前，读写 CSV 文件需要引入第三方库（如 `spark-csv`），并通过 Maven 添加依赖。 - 从 Spark 2.0 开始，CSV 数据源已经内置到 Spark SQL 中，因此可以直接使用 `format("csv")` 进行操作。 - 如果 CSV 文件较大，建议调整 `inferSchema` 参数以提高性能。若不需要自动推断数据类型，可以将其设置为 `"false"`。 ---