git.net

[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

help with right transform to read tgz file


Hi,

I am newbie to apache beam 
I am trying to write a simple pipeline using apache beam java sdk.
the pipleline will read a bunch of tgz files.
each tgz files have multiple CSV files with data

public static final void main(String args[]) throws Exception {

    PipelineOptions options = PipelineOptionsFactory.create();
    Pipeline pipeline = Pipeline.create(options);

    
    PCollection<MatchResult.Metadata> matches =
        pipeline.apply(FileIO.match().filepattern("/tmp/beam5/*.tgz"));
    
    PCollection<FileIO.ReadableFile> compGz = matches.apply(FileIO.readMatches().withCompression(Compression.GZIP));
    PCollection<String> contents = compGz.apply(FlatMapElements
            // uses imports from TypeDescriptors
            .into(TypeDescriptors.strings())
            .via((ReadableFile f) -> {
try {
return Arrays.asList(f.readFullyAsUTF8String().replaceAll("^@","").split("\\r?\\n|\\r", -1));
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
return null;
}) );
    PDone ret = contents.apply(TextIO.write().to("/tmp/beam6/output.txt").withoutSharding());

}

instead of returning flat list of strings, i tried parsing the f.readFullyAsUTF8String() and make CSVFileBean, but it does not seem to like

basically the above program is crude

i am looking for suggestions on right transform to transform this tgz into individual CSV bean POJO's that have name of CSV and contents

i am stuck decoding the tgz from readFullyAsUTF8String()

eventually i need to take each CSV bean and combine them

Eg.  test1.tgz has  foo_time1.csv, bar_time1.csv and 
       test2.tgz has  foo_time2.csv, bar_time2.csv

so i need to extract these CSV's and combine all the foo's and bar's

and possibly manipulate foo's, bar's by adding columns and transforming and then sending out to destinations which can be filesystem or kafka

thanks
Any help is appreciated
Sri