[jira] [Updated] (BEAM-12864) FlinkRunner (DOCKER) changes HDFS HOST to a name resolvable to host, rather than Docker

["Adam Pearce (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/BEAM-12864?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Adam Pearce updated BEAM-12864:
-------------------------------
    Resolution: Fixed
        Status: Resolved  (was: Triage Needed)

The source of this issue was the Hadoop HDFS core-site.xml configuration was set to "localhost" which appended "localhost:9000" to the access URL when Beam makes the connection.

Reconfiguring HDFS to use the IP of my Docker Host allowed Beam to access HDFS from within the Pipeline on the FlinkRunner.

 

> FlinkRunner (DOCKER) changes HDFS HOST to a name resolvable to host, rather than Docker
> ---------------------------------------------------------------------------------------
>
>                 Key: BEAM-12864
>                 URL: https://issues.apache.org/jira/browse/BEAM-12864
>             Project: Beam
>          Issue Type: Bug
>          Components: runner-flink
>    Affects Versions: 2.31.0
>         Environment: macOS 11.5.2, Flink 1.13.2, Beam 2.31.0, Docker for Mac 20.10.8
>            Reporter: Adam Pearce
>            Priority: P2
>
> I'm attempting to create a simple example of reading from an HDFS FileSystem using the Python SDK. I am able to do this with the direct runner, and am even able to read the filesystem directly in a simple Python file outside of a Beam pipeline (but using the Beam IO FileSystem class).
> When I create a Beam pipeline and submit it to Flink, it is unable to resolve the hostname of the Docker Host, because it is set to 'localhost'. I've tried setting `hdfs_host` in the pipeline options with the typical value of `host.docker.internal` to reach the Host's network, and even the IP address of my Docker Host (macOS) (which usually works and is resolvable when testing with dummy containers). The `host.docker.internal` fails because it is not resolvable by the Host. A situation is created where the Host _and_ the container both need to be able to resolve the `hdfs_host` hostname.
> When using the IP, this is possible, but I believe that in preparation for the Flink run, Beam replaces the HDFS Host entry with "localhost" because that is what the IP resolves to on the Docker Host, which is then not resolvable by the Docker container.
> Users need to be able to explicitly set the HDFS Host parameter with respect to the Docker environment that the FlinkRunner executes the pipeline in, regardless of if the Host can resolve that hostname. In some cases, this could be another Docker container on the Docker network that is resolvable to the Docker container, but not to the Docker Host. Setting the "hdfs_host" to an IP Address should not result in replacement with "localhost".
> To summarize, running a Beam pipeline with the FlinkRunner (using the Docker environment), is unable to reach the Docker Host via networking, and therefor would be unable to connect to an HDFS Filesystem located on the Docker Host.
>  
> Code example:
>  
> {code:java}
> HDFS_HOSTNAME = '192.168.1.11' # Docker Host IP Address (macOS)
> HDFS_PORT = 9870
> hdfs_client_options = HadoopFileSystemOptions(hdfs_host=HDFS_HOSTNAME, hdfs_port=HDFS_PORT, hdfs_user="apearce")
> hdfs_filesystem = HadoopFileSystem(hdfs_client_options)
> input_file_hdfs = "hdfs://user/apearce/testdata/001.csv"
> # This works
> # for x in hdfs_filesystem.open(input_file_hdfs).readlines():
> #    print(x)
> p = beam.Pipeline(options=PipelineOptions())
> def run(argv=None, save_main_session=True):
>  config = {
>  "runner": "FlinkRunner",
>  "flink_master": "localhost:8081",
>  "environment_type": "DOCKER",
>  "save_main_session": True,
>  "hdfs_host": HDFS_HOSTNAME,
>  "hdfs_port": HDFS_PORT,
>  "hdfs_user": "apearce",
>  }
>  pipeline_options = PipelineOptions.from_dictionary(config)
>  with beam.Pipeline(options=pipeline_options) as p:
>    (
>      p
>      | 'ReadFromHDFS' >> beam.io.ReadFromText(input_file_hdfs)
>      | 'Print' >> beam.Map(print)
>    )
> if __name__ == '__main__':
>  logging.getLogger().setLevel(logging.INFO)
>  run(){code}
>  
>  



--
This message was sent by Atlassian Jira
(v8.3.4#803005)