google cloud dataflow can run locally but can't run on the cloud
<p>I can't run my pipeline locally, using the "DirectPipelineRunner" but when I use "BlockingDataflowPipelineRunner" to run on the cloud it always shows that "Failed to split source". I don't where is the problem of my pipeline.</p>

<p>I define my custom sources to read a lot of tgz file in my bucket. The following is my code.</p>

<pre><code>class My_Compressed_Source(filebasedsource.FileBasedSource):
    def read_records(self, file_name, range_tracker):
        import tarfile
        import re    
        start_offset = range_tracker.start_position()
        with self.open_file(file_name) as f:
            tar = tarfile.open(fileobj = f)            
            for member in tar.getmembers():
            f = tar.extractfile(member)
            content = f.read()                    
            # do some regular expression process to content 
</code></pre>

<p>And here is my pipeline</p>

<pre><code>pcoll = p | 'Read' &gt;&gt; beam.Read(My_Compressed_Source(args.input, splittable = False))
pcoll | beam.ParDo('parse tar member', Parse_Members())
p.run()
</code></pre>

<p>also, my input path of bucket is "--input gs://mybucket/source/*.tgz" and my job-id is "2016-10-14_09_21_43-4409773810679481086".
I am wondering should I set "splittable" to True. Or if there is something wrong. </p>

<p>Thanks for all the comments above.<br>
I tried several ways to solve my problem. And I have a little progress. I think it is necessary that I should describe my problem again.<br>
<br>
I add the compression_type argument and set it to <code>fileio.CompressionTypes.GZIP</code>. Just as below.</p>

<pre><code>pcoll = p | "Read" &gt;&gt; beam.io.Read(My_Compressed_Source(args.input, split = False, compression_type = fileio.CompressionTypes.GZIP))
pcoll | beam.ParDo('parse tar member', Parse_Members())
p.run()
</code></pre>

<p>I try to execute it locally through "DirectPipelineRunner", and tarfile.open() showed an error. So I modified my read_records function to the below version.</p>

<pre><code>class My_Compressed_Source(filebasedsource.FileBasedSource):
    def read_records(self, file_name, range_tracker):
    import tarfile
    import re
    import StringIO
    start_offset = range_tracker.start_position()
    f = self.open_file(file_name)
    start = range_tracker.start_position
    text_str = str(f.read(f._read_size))
    tar = tarfile.open(fileobj=StringIO.StringIO(text_str), bufsize=f._read_size)
    for mem in tar.getmembers():
        print mem.name
</code></pre>

<p>I specify the number of bytes for f.read() to get the whole content of my tgz file. And now the files are just tar. <br>
And then I use StringIO module to create a temporary file_obj to put in tarfile.open. Finally I used a loop to access each name of the file. I indeed extract some file name but just a few of files are accessed.<br></p>

<p>After trying different data source, I think the key is no matter which the source is, the length of f._read_size is always has a maximum 16384 = 16k. I wonder If there is some where to set the maximum of buffer size can let me access whole content, or I should try another way?</p>

python google-cloud-dataflow