Python code for beams with hdfs support in Pipeline

Question

I am running the Sentiment example here for tensorflow transform. https://github.com/tensorflow/transform/blob/master/examples/sentiment_example.py

For fn ReadAndShuffleData() defined in line 78-98, is it possible that in a similar way I can load files but from HDFS, not GCS?

I have tried a whole day with several beam API (beams-2.8.0) but failed, and the most promising one I think is using beams.io.hadoopfilesystem. But this fn actually produces a python file-object and cannot be read in using beams.io.ReadFromText() in a beam pipeline.

I also passed in HadoopFileSystemPipelineOptions correctly. Anyone can show me a direction to solve the issue or a 2/3-line code snippets or a workaround? Thank you very much!

p.s. hadoop 2.7.7, beams 2.8 and data is loaded correctly.

I think I may lack some theoretical understandings here, any references will be appreciated!

Adrien Renaud Adrien Renaud · Accepted Answer · 2019-05-28T09:56:51

You might use the apache_beam.Create transform:

Init signature: beam.Create(self, values, reshuffle=True)

Docstring: A transform that creates a PCollection from an iterable.

import apache_beam as beam
from apache_beam.options.pipeline_options import HadoopFileSystemOptions
from apache_beam.io.hadoopfilesystem import HadoopFileSystem

HDFS_HOSTNAME = 'foo.hadoop.com'
HDFS_PORT = 50070
hdfs_client_options = HadoopFileSystemOptions(hdfs_host=HDFS_HOSTNAME, hdfs_port=HDFS_PORT, hdfs_user="foobar")
hdfs_client = HadoopFileSystem(hdfs_client_options)

input_file_hdfs = "hdfs://foo/bar.csv"
f = hdfs_client.open(input_file_hdfs)

p = beam.Pipeline(options=PipelineOptions())
lines = p | 'ReadMyFile' >> beam.Create(f)
res = lines | "WriteMyFile" >> beam.io.WriteToText("./bar", ".csv")
p.run()

Python code for beams with hdfs support in Pipeline

1 Answers