GBK格式内容转码
val logRdd = sc.hadoopFile(logPath, classOf[TextInputFormat], classOf[LongWritable], classOf[Text])
.map(p => new String(p._2.getBytes, 0, p._2.getLength, "GBK"))
获取文件路径或文件名
hadoop mr 中使用如下方式
InputSplit inputSplit=(InputSplit)context.getInputSplit();
String filename=((FileSplit)inputSplit).getPath().getName();
spark 中
获取文件名中的ip信息，并将文件转码
val logRdd = sc.hadoopFile(logPath, classOf[TextInputFormat], classOf[LongWritable], classOf[Text])
.asInstanceOf[HadoopRDD[LongWritable,Text]]
.mapPartitionsWithInputSplit((inputSplit:InputSplit, iterator:Iterator[(LongWritable, Text)]) => {
val file = inputSplit.asInstanceOf[FileSplit]
val fileName = file.getPath.getName()
val reg = "((2(5[0-5]|[0-4]\\d))|[0-1]?\\d{1,2})(\\.((2(5[0-5]|[0-4]\\d))|[0-1]?\\d{1,2})){3}".r
val ip = reg.findFirstIn(fileName)
iterator.map(x => {ip.getOrElse("") + SEPRATOR + new String(x._2.getBytes, 0, x._2.getLength, "GBK")})
})
