Apache spark以正则表达式读取文件

Question

我正在向 HDFS 发送流并尝试使用 spark 读取文本文件。

JavaStreamingContext jssc = new JavaStreamingContext(jsc, new    
 Duration(1000));
JavaPairInputDStream<LongWritable, Text> textStream =   
jssc.fileStream("hdfs://myip:9000/travel/FlumeData.[0-9]*", 
LongWritable.class, Text.class, TextInputFormat.class);

在向 hdfs 发送流时，会创建一些 FlumeData.1234.tmp 文件，一旦收到完整数据，该文件就会转换为适当的文件，例如。 FlumeData.1234

我想忽略这个要读取的 .tmp 文件。我尝试使用正则表达式

hdfs://myip:9000/travel/FlumeData.[0-9]* hdfs://myip:9000/travel/FlumeData.//d*

但他们不工作。我正在寻找这样的东西 jssc.fileStream("hdfs://myip:9000/travel/FlumeData.[0-9]*", LongWritable.class, Text.class, TextInputFormat.class);

fileStream 不应从文件扩展名中读取 .tmp。

我还尝试按照 Hadoop 代码检索苍蝇列表

private  String pathValue(String PathVariable) throws IOException{



      Configuration conf = new Configuration();
      Path path = new Path(PathVariable);
      FileSystem fs = FileSystem.get(path.toUri(), conf);
      System.out.println("PathVariable" + fs.getWorkingDirectory());

      return fs.getName();
   } 

但它的文件系统对象 fs 没有文件名（）。由于新文件是在运行时创建的。我需要阅读他们创建的内容。

Answer 1

JavaPairInputDStream 重载的fileStream 方法带有过滤功能，我们可以编写一个过滤功能来过滤掉目录中的文件。

fileStream(directory, kClass, vClass, fClass, filter, newFilesOnly)

JavaPairInputDStream<LongWritable, Text> lines = jssc.fileStream("hdfs://myip:9000/travel/", LongWritable.class, Text.class, TextInputFormat.class, new Function<Path,Boolean> () {
        public Boolean call(Path path) throws Exception {
            System.out.println("Is path :"+path.getName());
            Pattern pattern =  Pattern.compile("FlumeData.[0-9]*");
            Matcher m = pattern.matcher(path.getName());
            System.out.println("Is path : " + path.getName().toString() + " matching "
                + " ? , " + m.matches());
            return  m.matches();
        }}, true);

请使用上面的代码运行，希望能解决问题。

Answer 2

您需要使用 () 选择器来选择可以不匹配的部分。如果不指定任何部分，则返回整个匹配项。

在你的情况下，如果我没有误解你想在你的例子中选择：

FlumeData.1234 from FlumeData.1234.tmp 

为此，您需要的简单正则表达式是：

(.*).tmp

如果您想选择 .tmp 扩展名之前的所有内容。