Подсчитать разделитель между столбцами в RDD или DataFrame и переместить строки в отдельные RDD или кадры данных.

Question

Я новичок в программировании и запускаю несколько файлов TSV.gz в RDD или Dataframes.Я хочу подсчитать количество вкладок между столбцами после загрузки и переместить строки данных в отдельные СДР или кадры данных на основе приведенного ниже условия.

Общее количество столбцов = 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)

Пожалуйста, дайте мне знать, если то же самое можно достичь

Спасибо.!

Answer 1

Если файл csv, всегда используйте кадры данных.Вы можете использовать df.columns.length, который дает количество столбцов в CSV-файле.Вот пример кода для CSV-файла с 8 столбцами.Вы можете изменить соответственно для 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)
    }

  }
}

Answer 2

Прежде всего, начиная с Spark 2.0, настоятельно рекомендуется придерживаться SparkSQL, если вам не требуется низкоуровневый доступ к СДР.Это не ваш случай, поэтому не стесняйтесь забывать 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)