如何在 R 中使用 Spark 读取固定宽度的文件答案

【问题标题】：How to read fixed width files using Spark in R如何在 R 中使用 Spark 读取固定宽度的文件
【发布时间】：2019-08-16 00:46:10
【问题描述】：

我需要将一个 10GB 的固定宽度文件读取到数据帧中。如何在 R 中使用 Spark？

假设我的文本数据如下：

text <- c("0001BRAjonh   ",
"0002USAmarina ",
"0003GBPcharles")

我希望前 4 个字符与数据框的“ID”列相关联；从字符 5-7 将与列“国家”相关联；并从字符 8-14 关联到“名称”列

如果数据集很小，我会使用函数 read.fwf，但事实并非如此。

我可以使用 sparklyr::spark_read_text 函数将文件作为文本文件读取。但我不知道如何正确地将文件的值归因于数据框。

【问题讨论】：

我基于 substring 和 selectExpr 在 Scala 中为我构建了一个类。首先，我将文本文件中的所有模式设置在具有五列的外部 Hive 表下：表名、列号、列名、列开始、列结束。每一列都被转换成一个相应的数组，需要一段时间来构建每一列的解析语句。即使它将在 Scala 中，希望我在答案部分为它起草一个代理 Spark 答案？
绝对是的，@afeldman。它肯定会帮助我弄清楚如何在 R 中做到这一点。谢谢。

标签： r apache-spark bigdata sparkr sparklyr

【解决方案1】：

编辑：忘了说子字符串从 1 开始，数组从 0 开始，因为原因。

浏览并添加我在上列中谈到的代码。

该过程是动态的，并且基于名为 Input_Table 的 Hive 表。该表有 5 列：Table_Name、Column_Name、Column_Ordinal_Position、Column_Start 和 Column_Length。它是外部的，因此任何用户都可以将任何文件更改、删除和删除到文件夹位置。我快速从头开始构建，不使用实际代码，一切都有意义吗？

#Call Input DataFrame and the Hive Table. For hive table we make sure to only take correct column as well as the columns in correct order.
val inputDF       = spark.read.format(recordFormat).option("header","false").load(folderLocation + "/" + tableName + "." + tableFormat).rdd.toDF("Odd_Long_Name")
val inputSchemaDF = spark.sql("select * from Input_Table where Table_Name = '" + tableName + "'").sort($"Column_Ordinal_Position")

#Build all the arrays from the columns, rdd to map to collect changes a dataframe col to a array of strings. In this format I can iterator through the column.
val columnNameArray    = inputSchemaDF.selectExpr("Column_Name").rdd.map(x=>x.mkString).collect
val columnStartArray   = inputSchemaDF.selectExpr("Column_Start_Position").rdd.map(x=>x.mkString).collect
val columnLengthArray  = inputSchemaDF.selectExpr("Column_Length").rdd.map(x=>x.mkString).collect

#Make the iteraros as well as other variables that are meant to be overwritten
var columnAllocationIterator = 1
var localCommand             = ""
var commandArray             = Array("") 

#Loop as there are as many columns in input table
while (columnAllocationIterator <= columnNameArray.length) {
  #overwrite the string command with the new command, thought odd long name was too accurate to not place into the code
  localCommand = "substring(Odd_Long_Name, " + columnStartArray(columnAllocationIterator-1) + ", " + columnLengthArray(columnAllocationIterator-1) + ") as " + columnNameArray(columnAllocationIterator-1) 

  #If the code is running the first time it overwrites the command array, else it just appends
  if (columnAllocationIterator==1) {
    commandArray = Array(localCommand)
  } else {
    commandArray = commandArray ++ Array(localCommand)
  }

  #I really like iterating my iterators like this
  columnAllocationIterator = columnAllocationIterator + 1
}

#Run all elements of the string array indepently against the table
val finalDF = inputDF.selectExpr(commandArray:_*)

【讨论】：