spark多目录输出

曾经终败给现在 2023-07-17 02:52 7阅读 0赞

### 背景 ###

现在有这么一个需求，spark 读取日志信息，日志中有info,error,debug 相关的日志，我们要将其读取并分别放到不同的目录中。当然了，每次读取落地一次也可以，但是这样相当于要读取多次，在文件很大的时候，效率这块肯定不好

### 解决 ###

**spark可以支持多目录输出，具体方法如下**

*  1、我们可以使用saveAsHadoopFile方法，并自定义一个类继承MultipleTextOutputFormat，代码如下

class RDDMultipleTextOutputFormat extends MultipleTextOutputFormat[Any, Any] {
      override def generateFileNameForKeyValue(key: Any, value: Any, name: String): String =
        (key + "/" + name)
    }

重写generateFileNameForKeyValue这个方法来自定义文件名称和路径，比如如果想写到文件中，那么就不要添加"/“这个参数，写到相应的目录下，通过添加”/指定"。

*  2、调用saveAsHadoopFile，代码如下

val sc = new SparkContext(conf)
        sc.textFile("/user/zgh/ceshi.log")
          .map(value => {
            if (value.contains("info")){
              ("info", value)
            }else if((value.contains("debug"))) {
              ("debug", value)
            } else if (value.contains("error")) {
              ("error", value)
            }else {("null", value)}
          } )
          .partitionBy(new HashPartitioner(3))
          .saveAsHadoopFile("/user/zgh/test", classOf[String], classOf[String],
            classOf[RDDMultipleTextOutputFormat])
        sc.stop()

**这里需要注意两点**  
1、.partitionBy(new HashPartitioner(3)) 这里设定分区的含义是为了让输入文件只为一个。因为有info,debug  
,error 3中类型么。如果不设定的话，分区默认数为2，多路输出的分区也为2 ，如果输出到目录的话，就会生成多个文件。  
2、saveAsHadoopFile partitionBy 这些方法都是pairRDD的方法，所以只能使用pairRDD来调用。如果不清楚这个方法到底属不属于pairRDD，可以直接点进方法，如下图（这样可以清晰显示）  
![image.png][]

[image.png]: /images/20230528/1f551f2bf9c24e75b6a0c63de877ef21.png