osdir.com

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

Re: help with right transform to read tgz file


The general approach in your example looks reasonable to me. I don't think there's anything built in to Beam to help with parsing the tar file format and I don't know how robust the method of replacing "^@" and then splitting on newlines will be. I'd likely use Apache's commons-compress library for walking through the bytes of the tar file, pulling out the file names and associated contents.

You should be able to put all of that logic into a single FlatMapElements invocation as in your example. I'd suggest returning KV<String, String> where the key is the file name (or perhaps just the prefix of the file name if that's what you want to combine on) and the value is the line of content.

It looks like Beam does include some capabilities for parsing CSV lines, but I have no experience using them. It looks like they're building upon Apache commons-csv, so you might consider using that library directly if you need to parse fields out of the CSV and do transformations.

On Fri, Dec 21, 2018 at 6:24 PM Sridevi Nookala <snookala@xxxxxxxxxxxxxxxxxxxx> wrote:

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