How can I get the file-name list of a directory from hdfs in pyspark?

You can use HDFS (or any other compatible Hadoop filesystem) API in pyspark with a little py4j magic. To list files from a specific directory use:

path = "/here/is/my/dir/"
fs = spark._jvm.org.apache.hadoop.fs.FileSystem.get(spark._jsc.hadoopConfiguration())
list_status = fs.listStatus(spark._jvm.org.apache.hadoop.fs.Path(path))
result = [file.getPath().getName() for file in list_status]

The elements of list_status collection are of type FileSystem. With this API you can get files metadata, such as information if it's directory, mode, owner, group, acls and use these information to filter out unwanted files.

There is no functionality in Pyspark for this (EDIT: see answer by Mariusz and UPDATE at the end) - this functionality is provided in the Python package pywebhdfs (simply install by pip install pywebhdfs):

from pywebhdfs.webhdfs import PyWebHdfsClient
from pprint import pprint

hdfs = PyWebHdfsClient(host='192.10.10.73',port='50070', user_name='ctsats')  # your Namenode IP & username here
my_dir = 'user/ctsats'
pprint(hdfs.list_dir(my_dir))

The result is a (rather long) Python dictionary (not shown) - experiment a little to get a feeling. You can parse it to get the names and types (file/directory) as follows:

data = hdfs.list_dir(my_dir)
dd = [[x["pathSuffix"], x["type"]] for x in data["FileStatuses"]["FileStatus"]]
dd
# [[u'.Trash', u'DIRECTORY'], [u'.sparkStaging', u'DIRECTORY'], [u'checkpoint', u'DIRECTORY'], [u'datathon', u'DIRECTORY'], [u'ms-spark', u'DIRECTORY'], [u'projects', u'DIRECTORY'], [u'recsys', u'DIRECTORY'], [u'sparklyr', u'DIRECTORY'], [u'test.data', u'FILE'], [u'word2vec', u'DIRECTORY']]

From here, a simple list comprehension should do the job - for example, in my case where I have both files & directories present, here is how I coud keep only the directories:

sub_dirs = [x[0] for x in dd if x[1]=='DIRECTORY']
sub_dirs
# [u'.Trash', u'.sparkStaging', u'checkpoint', u'datathon', u'ms-spark', u'projects', u'recsys', u'sparklyr', u'word2vec']

For comparison, here is the actual listing of the same directory:

[ctsats@dev-hd-01 ~]$ hadoop fs -ls
Found 10 items
drwx------   - ctsats supergroup          0 2016-06-08 13:31 .Trash
drwxr-xr-x   - ctsats supergroup          0 2016-12-15 20:18 .sparkStaging
drwxr-xr-x   - ctsats supergroup          0 2016-06-23 13:23 checkpoint
drwxr-xr-x   - ctsats supergroup          0 2016-02-03 15:40 datathon
drwxr-xr-x   - ctsats supergroup          0 2016-04-25 10:56 ms-spark
drwxr-xr-x   - ctsats supergroup          0 2016-06-30 15:51 projects
drwxr-xr-x   - ctsats supergroup          0 2016-04-14 18:55 recsys
drwxr-xr-x   - ctsats supergroup          0 2016-11-07 12:46 sparklyr
-rw-r--r--   3 ctsats supergroup         90 2016-02-03 16:55 test.data
drwxr-xr-x   - ctsats supergroup          0 2016-12-15 20:18 word2vec

The WebHDFS service in your Hadoop cluster must be enabled, i.e. your hdfs-site.xml file must include the following entry:

<property>
    <name>dfs.webhdfs.enabled</name>
    <value>true</value>
</property>

UPDATE (after Mariusz's answer): Here is an adaptation of Mariusz's answer for Spark 1.6 (you need to replace spark with sc) in my example:

path="/user/ctsats"
fs = sc._jvm.org.apache.hadoop.fs.FileSystem.get(sc._jsc.hadoopConfiguration())
list_status = fs.listStatus(sc._jvm.org.apache.hadoop.fs.Path(path))
result = [file.getPath().getName() for file in list_status]
result
# [u'.Trash', u'.sparkStaging', u'checkpoint', u'datathon', u'ms-spark', u'projects', u'recsys', u'sparklyr', u'test.data', u'word2vec']

The issue here is that both files and subfolders are returned, without any means for distinction between them. The pywebhdfs solution, as demonstrated, does not suffer from this...

I guess there are ways to overcome this, but you will have to dig into the py4j API - despite the deceiving appearence, list_status is not a Python list:

list_status
# JavaObject id=o40

for python 3 using subprocess library:

from subprocess import Popen, PIPE
hdfs_path = '/path/to/the/designated/folder'
process = Popen(f'hdfs dfs -ls -h {hdfs_path}', shell=True, stdout=PIPE, stderr=PIPE)
std_out, std_err = process.communicate()
list_of_file_names = [fn.split(' ')[-1].split('/')[-1] for fn in std_out.decode().readlines()[1:]][:-1]
list_of_file_names_with_full_address = [fn.split(' ')[-1] for fn in std_out.decode().readlines()[1:]][:-1]