Файл moveToHDFS не найден в скрипте PySpark

Question

У меня немного странная проблема.

Когда я запускаю эту команду в терминале, она работает и действительно копирует файл в нужное место.

hdfs dfs -copyFromLocal concsessions.csv /user/username/spark_exports/

Но,когда я запускаю его как часть моего скрипта (ниже), это не так, и я получаю эту ошибку - кто-нибудь может мне помочь?

Я не уверен, что я делаю неправильно, определенно что-то!

OSError: [Errno 2] No such file or directory

Код:

from pyspark.sql import SparkSession
from datetime import datetime

#Set the date for the filename
now = datetime.now()
yday  = long(now.strftime('%s')) - 24*60*60

spark = SparkSession\
.builder\
.appName('wap')\
.master('yarn')\
.enableHiveSupport()\
.getOrCreate()

import datetime
import pyspark.sql.functions as F
from pyspark.sql.functions import col

#The below prints your results to your chosen destination (Hive, Stdout, CSV)

print('data load starting...')

cmd = '''select * from db.conc_sessions'''
df1 = spark.sql(cmd)
df1.printSchema()
print('data ingested successfully')

print('setting variables...')

timestart= '2019-10-14 00:00:00'
timeend= '2019-10-14 23:59:59'
time_to_check = datetime.datetime.strptime(timestart, '%Y-%m-%d %H:%M:%S')

iters = 0
session = 0
add = []

print('begin iteration...')
while iters < 96:

    time_to_add = iters * 900
    time_to_checkx = time_to_check + datetime.timedelta(seconds=time_to_add)
    stringtime = time_to_checkx.strftime("%m/%d/%Y, %H:%M:%S")

    iters = iters + 1

    spark_date_format = "YYYY-MM-dd hh:mm:ss"
    df1 = df1.withColumn('start_timestamp', F.to_timestamp(df1.start_time, spark_date_format))
    df1 = df1.withColumn('end_timestamp', F.to_timestamp(df1.end_time, spark_date_format))

    filterx = df1.filter( (df1.start_time < time_to_checkx) & (df1.end_time > time_to_checkx ))

    session = filterx.count()
    newrow = [stringtime, session]
    add.append(newrow)

import pandas as pd
output = pd.DataFrame.from_records(add)
output.columns = ['time','count']
output = output.groupby(['time'])[['count']].agg('sum').reset_index()
output.to_csv('concsessions.csv', sep=',')

#copy the CSV from the local server to HDFS
import subprocess
subprocess.call("hdfs dfs -copyFromLocal concsessions.csv /user/username/spark_exports/")

Answer 1

Вы должны использовать shell=True в subprocess.call.

From https://docs.python.org/2/library/subprocess.html#frequently-used-arguments

Аргумент оболочки (по умолчанию False) указывает, использовать ли оболочку в качестве программы для выполнения,Если shell равен True, рекомендуется передавать аргументы как строку, а не как последовательность.

В Unix с shell = True оболочкой по умолчанию является /bin/sh.

Редактироватьпозвоните, как показано ниже, и посмотрите.

subprocess.call("hdfs dfs -copyFromLocal concsessions.csv /user/username/spark_exports/", shell=True)

Answer 2

subprocess.call(['hdfs',  'dfs',  '-copyFromLocal' , 'concession.csv', 'user/username/spark_exports'])