下面的示例读取.csv 文件，格式与问题中的格式相同。
有一些细节我想先解释一下。
在表架构中，字段：rec_value: Decimal(2,1) 必须是 rec_value: Decimal(3,1)，原因如下：
DECIMAL 类型表示具有固定precision 和scale 的数字。
创建DECIMAL 列时，指定precision, p 和scale, s。
Precision 是总位数，与小数点的位置无关。
Scale 是小数点后的位数。
要在不损失精度的情况下表示数字 10.0，您需要一个
DECIMAL 类型，precision 至少为 3，scale 至少为 1。
所以Hive 表将是：
CREATE TABLE tab_data (
rec_id INT,
rec_name STRING,
rec_value DECIMAL(3,1),
rec_created TIMESTAMP
) STORED AS PARQUET;
完整的 Scala 代码
import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.types.{DataTypes, IntegerType, StringType, StructField, StructType, TimestampType}
object CsvToParquet {
val spark = SparkSession
.builder()
.appName("CsvToParquet")
.master("local[*]")
.config("spark.sql.shuffle.partitions","200") //Change to a more reasonable default number of partitions for our data
.config("spark.sql.parquet.writeLegacyFormat", true) // To avoid issues with data type between Spark and Hive
// The convention used by Spark to write Parquet data is configurable.
// This is determined by the property spark.sql.parquet.writeLegacyFormat
// The default value is false. If set to "true",
// Spark will use the same convention as Hive for writing the Parquet data.
.getOrCreate()
val sc = spark.sparkContext
val inputPath = "hdfs://host:port/user/...../..../tab_data.csv"
val outputPath = "hdfs://host:port/user/hive/warehouse/test.db/tab_data"
def main(args: Array[String]): Unit = {
Logger.getRootLogger.setLevel(Level.ERROR)
try {
val DecimalType = DataTypes.createDecimalType(3, 1)
/**
* the data schema
*/
val schema = StructType(List(StructField("rec_id", IntegerType, true), StructField("rec_name",StringType, true),
StructField("rec_value",DecimalType),StructField("rec_created",TimestampType, true)))
/**
* Reading the data from HDFS as .csv text file
*/
val data = spark
.read
.option("sep","|")
.option("timestampFormat","yyyy-MM-dd HH:mm:ss.S")
.option("inferSchema",false)
.schema(schema)
.csv(inputPath)
data.show(truncate = false)
data.schema.printTreeString()
/**
* Writing the data as Parquet file
*/
data
.write
.mode(SaveMode.Append)
.option("compression", "none") // Assuming no data compression
.parquet(outputPath)
} finally {
sc.stop()
println("SparkContext stopped")
spark.stop()
println("SparkSession stopped")
}
}
}
输入文件为.csv 制表符分隔字段
10|customer1|10.0|2016-09-07 08:38:00.0
20|customer2|24.0|2016-09-08 10:45:00.0
30|customer3|35.0|2016-09-10 03:26:00.0
40|customer1|46.0|2016-09-11 08:38:00.0
........
读自Spark
+------+---------+---------+-------------------+
|rec_id|rec_name |rec_value|rec_created |
+------+---------+---------+-------------------+
|10 |customer1|10.0 |2016-09-07 08:38:00|
|20 |customer2|24.0 |2016-09-08 10:45:00|
|30 |customer3|35.0 |2016-09-10 03:26:00|
|40 |customer1|46.0 |2016-09-11 08:38:00|
......
架构
root
|-- rec_id: integer (nullable = true)
|-- rec_name: string (nullable = true)
|-- rec_value: decimal(3,1) (nullable = true)
|-- rec_created: timestamp (nullable = true)
阅读自Hive
SELECT *
FROM tab_data;
+------------------+--------------------+---------------------+------------------------+--+
| tab_data.rec_id | tab_data.rec_name | tab_data.rec_value | tab_data.rec_created |
+------------------+--------------------+---------------------+------------------------+--+
| 10 | customer1 | 10 | 2016-09-07 08:38:00.0 |
| 20 | customer2 | 24 | 2016-09-08 10:45:00.0 |
| 30 | customer3 | 35 | 2016-09-10 03:26:00.0 |
| 40 | customer1 | 46 | 2016-09-11 08:38:00.0 |
.....
希望这会有所帮助。