java RDD获取每个元素的类型和索引

piv4azn7  于 2022-12-25  发布在  Java
关注(0)|答案(1)|浏览(133)

我正在寻找获取Java中每个元素的类型和索引的方法。

['0,1,hi,1.0', '2,3,String,String2', '1.0,2.0,3,String']

那我想

[(0, int),(1, int),(2, String),(3, Double),(0, int) ........]

这样我就可以通过键来减少并且看到每一列的数据类型。我在python中实现了它,但是不确定如何在Java中实现它。有什么方法可以做到这一点吗?下面是我如何在python中实现它的

def infer_type(partition):
for row in partition:
    value = ""
    idx = 0
    for i in range(len(row)):
        if row[0] == self.prop.comment:
            break
        if row[i] == self.prop.delimiter or i == (len(row) - 1):
            if i == len(row) - 1:
                value += str(row[i])
            if bool(value.strip()) == False:
                yield (idx, 'None')
            elif int_regex_match.match(value):
                yield (idx, 'int')
            elif float_regex_match.match(value):
                yield (idx, 'float')
            else:
                if date_regex_match.match(value):
                    yield (idx, 'date')
                else:
                    yield (idx, 'str')
                idx += 1
                value = ""
        else:
            value += str(row[i])
    rdd = rdd.mapPartitions(infer_type).map(lambda x: ((x[0], x[1]), 1)).reduceByKey(add).map(
    lambda x: (x[0][0], (x[0][1], x[1])))

编辑:这就是我现在得到的。但是,我不能得到元组的迭代器。

PairFlatMapFunction map = new PairFlatMapFunction<Iterator<String>, Integer, String>(){

        @Override
        public Iterator<Tuple2<Integer, String>> call(Iterator<String> iterator) throws Exception {
            // TODO Auto-generated method stub
            while(iterator.hasNext()) {
                String[] row = iterator.next().split(",");
                for(int j = 0; j<row.length;j++) {
                    if(row[j].matches(int_regex)) {
                        Tuple2<Integer, String> result =new Tuple2(j, "int");
                        // return iterator of result..?
                    }else if(row[j].matches(float_regex)) {
                        Tuple2<Integer, String> result =new Tuple2(j, "float");
                        // return iterator of result..?
                    }else if(row[j].matches(date_regex_match)) {
                        Tuple2<Integer, String> result =new Tuple2(j, "date");
                        // return iterator of result..?
                    }else {
                        Tuple2<Integer, String> result =new Tuple2(j, "str");
                        // return iterator of result..?
                    }
                }
            }
        }
 };
JavaPairRDD pair_rdd = rdd.mapPartitionsToPair(map, false);
luaexgnf

luaexgnf1#

根据你表达的需求,我不明白你为什么要用mapPartition而不是简单的map,还有一个错误是你应该用flatMapToPair而不是mapToPair
为了达到你的目的,你的flatmap函数需要将一个字符串(例如"0,1,hi,1.0")Map到一个元组的迭代器,为此,你可以简单地创建一个ArrayList的计算结果:

@Override
public Iterator<Tuple2<Integer, String>> call(String row) throws Exception {
    String[] split_row = row.split(",");
    //create list
    List<Tuple2<Integer, String>> result = new ArrayList<>()
    for(int j = 0; j<split_row.length;j++) {
        if(split_row[j].matches(int_regex)) {
            result.add(new Tuple2(j, "int"));
        } //else ...
    }
    //return the iterator
    return result.iterator();
}

如果您实际上需要使用mapPartition,则可以将相同的逻辑应用于您的函数。

相关问题