计算 spark RDD 或 DataFrame 中列之间的分隔符并将行移动到单独的 RDD 或数据帧答案

【问题标题】：To count the delimiter between columns in spark RDD or DataFrame and move the rows to separate RDD or dataframes计算 spark RDD 或 DataFrame 中列之间的分隔符并将行移动到单独的 RDD 或数据帧
【发布时间】：2019-02-06 23:35:08
【问题描述】：

我是 spark 编程的新手，我正在将多个 TSV.gz 文件加载到 RDD 或 Dataframes 中。我想计算加载后列之间的选项卡数，并根据以下条件将数据行移动到单独的 RDD 或数据帧。

总列数 = 996

If the number of tab counts = 995 -> move to another RDD or DF
If the number of tab counts < 995 -> move to another RDD or DF
If the number of tab counts > 995 -> move to another RDD or DF

我尝试了以下但返回布尔值

val textFile = sc.textFile("/abc/*.tsv.gz")
textFile.map(line=>line.split("\t"))
val file1 = textFile.filter(line => line.contains("\t").count() > 995)
val file2 = textFile.filter(line => line.contains("\t").count() < 995)

如果能达到同样的效果，请告诉我

谢谢！

【问题讨论】：

标签： scala apache-spark pyspark apache-spark-sql

【解决方案1】：

首先，从 Spark 2.0 开始，强烈建议您坚持使用 SparkSQL，除非您需要对 RDD 进行一些低级访问。这不是你的情况，所以在你学习的时候可以暂时忘记 RDD。

您想要实现的目标可以通过多种方式完成。假设 TSV 具有标题行，或者您可以为列分配名称。利用 CSV 格式阅读器，只需使用 \t 作为分隔符：

val all = spark.read
  .format("csv")
  .option("header", "true")
  .option("delimiter", "\t")
  .option("inferSchema", "true")
  .load("file.tsv")

接下来假设 left 是第 994 列的名称，center 是第 995 列的名称，right 是第 996 列的名称。

val left = all.filter(col("center").isNull)
val center = all.filter(col("center").isNotNull && col("right").isNull)
val right = all.filter(col("right").isNotNull)

【讨论】：

【解决方案2】：

如果文件是 csv，请始终使用数据框。您可以使用 df.columns.length 给出 csv 文件中的列数。这是一个包含 8 列的 csv 文件的示例代码。您可以针对 996 列进行相应修改。

emp1.csv 的内容

7369     "SMITH"     "CLERK"     7902    "17-Dec-80"     800     20  10
7499     "ALLEN"     "SALESMAN"  7698    "20-Feb-81"     1600    300     30

火花码

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql._
object StackOverflow {
  def main(args: Array[String]) {
    Logger.getLogger("org").setLevel(Level.ERROR)
    val spark = SparkSession.builder().appName("Testing..").master("local[*]").getOrCreate()
    import spark.implicits._

    val emp1 = spark.read.format("csv").option("delimiter","\t").load("in/emp1.csv")
    emp1.show(false)
    val col_len = emp1.columns.length
    if(col_len == 8) {
      val df1 = emp1.toDF("empno", "ename", "job", "mgr", "hiredate", "sal", "comm", "deptno")
      df1.show(false)
    }
    if(col_len== 7) {
      val df2 = emp1.toDF("empno", "ename", "job", "mgr", "hiredate", "sal", "comm")
      df2.show(false)
    }

  }
}

【讨论】：