如何检查 HDFS 文件夹是否包含 CSV parquet 文件？ [复制]答案

【问题标题】：How to check if HDFS folder contain CSV parquet files? [duplicate]如何检查 HDFS 文件夹是否包含 CSV parquet 文件？ [复制]
【发布时间】：2020-10-06 14:31:08
【问题描述】：

如何以编程方式检查何时使用

spark.read.csv(path)

或

spark.read.parquet(path)

无需用户说明路径是否包含镶木地板或文本文件。路径应该在 HDFS 上。

【问题讨论】：

whooowww 这是一个很酷的问题。我的意思是，这很复杂..您应该先查看文件格式..至少有扫描或其他内容可以告诉您文件是哪种格式
您的问题与stackoverflow.com/questions/33394884/…高度相似
@Nick，您在链接中提出的解决方案是一个非常好的方法

标签： scala apache-spark hdfs

【解决方案1】：

我将利用 scala Try 并尝试使用 orElse 函数一一导入文件类型，而不是以编程方式检查其扩展名 -

def readCsv(): Try[DataFrame] = ???
def readParquet(): Try[DataFrame] = ???

val dfTry: Try[DataFrame] = readCsv().orElse(readParquet())

如果您有更多 parquet 读取请求，可以先拨打 readParquet() 电话。

【讨论】：

是 try ... catch 异常处理吗？如果是这样，那么这不是一个好方法。我更喜欢 if/else 方法来提高性能。
它可以工作，但是开销很大。

【解决方案2】：

一种方法是运行hdfs dfs -ls 命令并检查输出以查看输入目录是否包含csv 或parquet 文件。

举个例子

// This two imports are necessary to run shell commands from Scala
import scala.sys.process._
import scala.language.postfixOps 

// As a little example of how it could be
def getExtension(s: String): String = {
    if(s.contains(".parquet")) "parquet"
    else if(s.contains(".csv")) "csv"
    else if (s.contains(".txt")) "txt"
    else "unknown"
}

val inputDirCsv = "hdfs://quickstart.cloudera:8020/user/cloudera/csv"

val inputDirParquet = "hdfs://quickstart.cloudera:8020/user/cloudera/parquet"

val lsCommand = Seq("hdfs", "dfs", "-ls", inputDirCsv).!!
println(lsCommand)
/*
Found 4 items
-rw-r--r--   1 cloudera supergroup        109 2020-10-06 06:40 hdfs://quickstart.cloudera:8020/user/cloudera/csv/EmployeeManager.csv
-rw-r--r--   1 cloudera supergroup       8754 2020-10-06 06:40 hdfs://quickstart.cloudera:8020/user/cloudera/csv/amigos.csv
-rw-r--r--   1 cloudera supergroup        142 2020-10-06 06:40 hdfs://quickstart.cloudera:8020/user/cloudera/csv/updated_departments.csv
-rw-r--r--   1 cloudera supergroup         79 2020-10-06 06:40 hdfs://quickstart.cloudera:8020/user/cloudera/csv/user.csv
 */
println(getExtension(lsCommand)) // csv

val lsCommand1 = Seq("hdfs", "dfs", "-ls", inputDirParquet).!!
println(lsCommand1)
/*
Found 5 items
-rw-r--r--   3 cloudera supergroup          0 2020-04-24 22:28 hdfs://quickstart.cloudera:8020/user/cloudera/parquet/_SUCCESS
-rw-r--r--   3 cloudera supergroup        599 2020-04-24 22:28 hdfs://quickstart.cloudera:8020/user/cloudera/parquet/part-00000-ad9007ac-c3a8-45b1-bad3-fb608c759303-c000.snappy.parquet
-rw-r--r--   3 cloudera supergroup        645 2020-04-24 22:28 hdfs://quickstart.cloudera:8020/user/cloudera/parquet/part-00001-ad9007ac-c3a8-45b1-bad3-fb608c759303-c000.snappy.parquet
-rw-r--r--   3 cloudera supergroup        586 2020-04-24 22:28 hdfs://quickstart.cloudera:8020/user/cloudera/parquet/part-00002-ad9007ac-c3a8-45b1-bad3-fb608c759303-c000.snappy.parquet
-rw-r--r--   3 cloudera supergroup        645 2020-04-24 22:28 hdfs://quickstart.cloudera:8020/user/cloudera/parquet/part-00003-ad9007ac-c3a8-45b1-bad3-fb608c759303-c000.snappy.parquet
 */
println(getExtension(lsCommand1)) // parquet

【讨论】：

嗯，这是一个很酷的方法，但我担心他是否可以在 scala 中调用执行
如果hdfs cli的服务器版本改变了，输出的格式也随之改变了怎么办？最好使用 scala 库来执行此操作，而不是依赖命令行。