拆分数据数据类型后的Spark RDD如何在不更改数据类型的情况下拆分

Posted 2023-04-15

【中文标题】拆分数据数据类型后的Spark RDD如何在不更改数据类型的情况下拆分

【英文标题】：Spark RDD after splitting of data data type is changed how can i split without changing data type

【发布时间】：2020-10-22 07:38:50

【问题描述】：

在拆分数据数据类型更改后，我已将数据从文本文件加载到 Spark RDD。如何在不更改数据类型的情况下进行拆分，或者如何将拆分后的数据转换为原始数据类型？

我的代码

from pyspark import SparkConf, SparkContext 
conf = SparkConf().setMaster("local").setAppName("Movie") 
sc = SparkContext(conf = conf) 
movies = sc.textFile("file:///SaprkCourse/movie/movies.txt")
data=movies.map(lambda x: x.split(","))
data.collect()

我的意见是这样的

userId,movieId,rating,timestamp
1,1,4.0,964982703
1,3,4.0,964981247
1,6,4.0,964982224
1,47,5.0,964983815
1,50,5.0,964982931

拆分后我的完整数据改为String类型

我要求输出与输入文本文件中的数据类型相同，如IntegerType, IntegerType, IntegerType, IntegerType

【参考方案1】：

读取文本文件时的火花会影响所有列的 StringType 类型，因此如果要将列视为 IntegerType，则需要强制转换它们。

【参考方案2】：

你的数据是 csv， 您应该使用 sparkSession，使用 csv 读取数据并定义您的架构。

scala 代码：

val schema = new Structype()
.add("userId",IntegerType)
.add("movieId",IntegerType)
.add("rating",IntegerType)
.add("timestamp",TimestampType)


spark.read.schema(schema).csv("file:///SaprkCourse/movie/movies.txt")

如果您想将文件作为文本继续阅读，您可以转换每一列：

斯卡拉：

import org.apache.spark.sql.functions.col
import org.apache.spark.sql.types.IntegerType,TimestampType

   val df = data
    .select(
    col("userId").cast(IntegerType),
    col("movieId").cast(IntegerType),
    col("rating").cast(IntegerType),
    col("timestamp").cast(TimestampType)
    )

【讨论】：

嗨，我的输入文件是文本文件，但不是 CSV

CSV 文件是文本，这个答案的重点是输入文本文件中的数据被组织为 CSV