如何使用 Spark 从 .sql 转储中提取包含数据的表？

Question

我有大约四个 *.sql 独立转储（每个大约 20GB），我需要将它们转换为 Apache Spark 中的数据集。

我尝试使用 InnoDB 安装和制作本地数据库并导入转储，但这似乎太慢（花了大约 10 个小时）

我直接将文件读入 spark 使用

import org.apache.spark.sql.SparkSession

var sparkSession = SparkSession.builder().appName("sparkSession").getOrCreate()
var myQueryFile = sc.textFile("C:/Users/some_db.sql")

//Convert this to indexed dataframe so you can parse multiple line create / data statements. 
//This will also show you the structure of the sql dump for your usecase.

var myQueryFileDF = myQueryFile.toDF.withColumn("index",monotonically_increasing_id()).withColumnRenamed("value","text") 


// Identify all tables and data in the sql dump along with their indexes

var tableStructures = myQueryFileDF.filter(col("text").contains("CREATE TABLE"))
var tableStructureEnds = myQueryFileDF.filter(col("text").contains(") ENGINE"))

println(" If there is a count mismatch between these values choose different substring "+ tableStructures.count()+ " " + tableStructureEnds.count())

var tableData = myQueryFileDF.filter(col("text").contains("INSERT INTO "))

问题在于转储包含多个表，每个表都需要成为数据集。为此，我需要了解我们是否可以为一张桌子做到这一点。有没有为 scala spark 编写的 .sql 解析器？

有没有更快的方法？我可以从 .sql 自包含文件中直接将其读入 hive 吗？

更新 1：我正在根据 Ajay 给出的输入为此编写解析器

更新 2：将所有内容更改为基于数据集的代码以按照建议使用 SQL 解析器

Answer 1

有没有为 scala spark 编写的 .sql 解析器？

是的，有一个，而且您似乎已经在使用它了。这就是 Spark SQL 本身！惊讶吗？

SQL 解析器接口 (ParserInterface) 可以从 SQL 语句的文本表示创建关系实体。这几乎是你的情况，不是吗？

请注意，ParserInterface 一次只处理一条 SQL 语句，因此您必须以某种方式解析整个转储并找到表定义和行。

ParserInterface 可用作 sqlParser 或 SessionState。

scala> :type spark
org.apache.spark.sql.SparkSession

scala> :type spark.sessionState.sqlParser
org.apache.spark.sql.catalyst.parser.ParserInterface

Spark SQL 提供了几种方法来提供接口的入口点，例如SparkSession.sql、Dataset.selectExpr 或简单的 expr 标准函数。您也可以直接使用 SQL 解析器。

---

无耻的插件您可能想在 Mastering Spark SQL 一书中了解ParserInterface — SQL Parser Contract。

Answer 2

你需要自己解析。它需要以下步骤 -

1. 为每个表创建一个类。
2. 使用 textFile 加载文件。
3. 过滤掉除插入语句之外的所有语句。
4. 然后根据插入语句中存在的表名，使用过滤器将 RDD 拆分为多个 RDD。
5. 对于每个 RDD，使用 map 解析插入语句中存在的值并创建对象。
6. 现在将 RDD 转换为数据集。