diff --git a/python/paddle/fluid/async_executor.py b/python/paddle/fluid/async_executor.py index 787a6a6b9e19e6253664187ee166609c2e2abc1f..cce7ec5cca1151adab2326094450f0bcb06ab504 100644 --- a/python/paddle/fluid/async_executor.py +++ b/python/paddle/fluid/async_executor.py @@ -25,6 +25,7 @@ from google.protobuf import text_format from . import io from .data_feed_desc import DataFeedDesc from .distributed import ps_instance +from .contrib.utils import hdfs_utils as hdfs __all__ = ['AsyncExecutor'] @@ -152,6 +153,22 @@ class AsyncExecutor(object): data_feed.desc(), filelist, thread_num, fetch_var_names, debug) + def download_data(self, afs_path, local_path, fs_default_name, ugi, process_num=12): + hadoop_home = "$HADOOP_HOME" + + configs = { + "fs.default.name": fs_default_name, + "hadoop.job.ugi": ugi + } + + client = hdfs.HDFSClient(hadoop_home, configs) + downloads = hdfs.multi_download( + client, + afs_path, + local_path, + self.instance.get_worker_index(), + self.instance.get_node_cnt() / 2, + multi_processes=process_num) def config_distributed_nodes(self, dist_opt): @@ -179,10 +196,11 @@ class AsyncExecutor(object): self.executor.gather_servers(ips, self.instance.get_node_cnt()) self.instance.barrier_all() #wait all worker start self.instance.barrier_all() #wait init model + self.instance.barrier_all() #wait for download_data self.instance.barrier_all() #wait worker do all things self.instance.barrier_all() #sync - def init_worker(self, dist_desc): + def init_worker(self, dist_desc, afs_path, local_path, fs_default_name, ugi): self.instance.barrier_all() #wait all server start ips = self.instance.gather_ips() self.executor.init_worker(dist_desc, ips, self.instance.get_node_cnt(), self.instance._rankid) @@ -190,6 +208,8 @@ class AsyncExecutor(object): if self.instance.is_first_worker(): self.executor.init_model() self.instance.barrier_all() #wait init model + self.download_data(afs_path, local_path, fs_default_name, ugi, process_num=12) + self.instance.barrier_all() #wait for download_data def init_model(self): self.executor.init_model() diff --git a/python/paddle/fluid/contrib/utils/__init__.py b/python/paddle/fluid/contrib/utils/__init__.py index 6e479bdc2b93c1189ba07a6f20b2408c34110b93..2fe9f702f3de8565ca8800fcea9d5424b1d7cee9 100644 --- a/python/paddle/fluid/contrib/utils/__init__.py +++ b/python/paddle/fluid/contrib/utils/__init__.py @@ -13,8 +13,8 @@ # limitations under the License. from __future__ import print_function -from . import lookup_table_utils -from .lookup_table_utils import * +#from . import lookup_table_utils +#from .lookup_table_utils import * from . import hdfs_utils from .hdfs_utils import *