Pyspark：HDFSパス上のファイル/ディレクトリのリストを取得

Question

タイトルのとおり。私はtextFileを認識していますが、名前が示唆するように、テキストファイルでのみ動作します。 HDFS（またはローカルパス）上のパス内のファイル/ディレクトリにアクセスする必要があります。あなたは、ディレクトリ内すべてファイルで読みたい場合は、私はpysparkに助け

Answer 1

Sparkは、データをロードするときにドメインを開始するデータ処理ツールとして考えるのが有益だと思います。それは多くのフォーマットを読むことができ、Hadoopのglob式をサポートしています。これはHDFSの複数のパスからの読み込みには非常に便利ですが、ディレクトリやファイルを走査するのに気づいている組み込み機能はありません。 HadoopまたはHDFSとの相互作用に特有のユーティリティ。

esutilとhdfsなど、必要な操作を行うツールがいくつかあります。 hdfs libはCLIとAPIの両方をサポートしていますので、「PythonでHDFSファイルをリストするにはどうすればいいですか」のように右に移動してくださいhere。それは次のようになります。JVMゲートウェイを使用して

from hdfs import Config 
client = Config().get_client('dev') 
files = client.list('the_dir_path') 

Answer 2

ため

感謝を使用しています、sc.wholeTextFiles[doc]をチェックアウトしますが、ファイルの内容は、単一の行の値に読み込まれることに注意してくださいこれはおそらく望ましい結果ではありません。

あなただけのいくつかのファイルを読みたい場合は、RDDにDataFrameから変換するように思えるその後、（通常のHDFSのlsコマンドを使用して、プラス何が必要フィルタリング）パスのリストを生成し、sqlContext.read.text[doc]にそれを渡すと、最良のアプローチ。

Answer 3

多分そうエレガントではありませんが、いくつかのケースでは、以下のコードは役に立つことができます

URI   = sc._gateway.jvm.java.net.URI 
Path   = sc._gateway.jvm.org.apache.hadoop.fs.Path 
FileSystem = sc._gateway.jvm.org.apache.hadoop.fs.FileSystem 
Configuration = sc._gateway.jvm.org.apache.hadoop.conf.Configuration 


fs = FileSystem.get(URI("hdfs://somehost:8020"), Configuration()) 

status = fs.listStatus(Path('/some_dir/yet_another_one_dir/')) 

for fileStatus in status: 
    print fileStatus.getPath() 

Answer 4

あなたがPySparkを使用する場合は、対話的にコマンドを実行することができます：

---

リスト選択したディレクトリのすべてのファイル：

hdfs dfs -ls <path>例：hdfs dfs -ls /user/path：

import os 
import subprocess 

cmd = 'hdfs dfs -ls /user/path'.split() 
files = subprocess.check_output(cmd).strip().split('\n') 
for path in files: 
    print path 

---

または選択したディレクトリ内のファイルを検索します。

hdfs dfs -find <path> -name <expression>例：hdfs dfs -find /user/path -name *.txt：

import os 
import subprocess 

cmd = 'hdfs dfs -find {} -name *.txt'.format(source_dir).split() 
files = subprocess.check_output(cmd).strip().split('\n') 
for path in files: 
    filename = path.split(os.path.sep)[-1].split('.txt')[0] 
    print path, filename 

Answer 5

これを行う簡単な方法がありますsnakebiteライブラリを使用して

from snakebite.client import Client 

hadoop_client = Client(HADOOP_HOST, HADOOP_PORT, use_trash=False) 

for x in hadoop_client.ls(['/']): 

...  print x