Python Spark: аргумент без ключевого слова после ключевого слова arg

Question

У меня есть файл .py в PySpark следующим образом

from pyspark.sql import HiveContext, SQLContext, SparkSession
from pyspark.sql import Row

spark = SparkSession \
.builder \
.appName("Python Spark SQL basic example") \
.config("spark.some.config.option", "some-value") \
.getOrCreate()
sc = spark.sparkContext
sqlContext = SQLContext(sc)
iot = sc.textFile("file:///home/meter/df_final.csv")
iot  = iot.map(lambda p: p.split(","))
header = iot.first()
iot_f = iot.filter(lambda p:p != header)
iot_f = iot_f.map(lambda p: Row(DeviceImei = str(p[0]), DeviceTimeStamp = 
p[1], Avg_PF=float(p[2]),Sum_PF=float(p[3]),FRQ=int(p[4]),THDVL1=float(p[5]),T ,MachineName = str(p[16]),HUM = int(p[17]),OLI = int(p[18]),GridStatus = bool(p[19]))).toDF()

hc = HiveContext(sc)

iot_f.write.format("orc").saveAsTable("iot")

Теперь, когда я отправляю эту работу как ./bin/spark-submit --master yarn/path/to/file/file.py, я получаю ошибку как non-keyword arg after keyword arg

Куда я скучаю? Я хочу создать улейную таблицу с именем 'iot'.

Answer 1

В примере кода вы передаете номер аргумента.

iot_f = iot_f.map(
          lambda p: Row(
              DeviceImei = str(p[0]), 
              DeviceTimeStamp = p[1], 
              Avg_PF=float(p[2]),
              Sum_PF=float(p[3]),
              FRQ=int(p[4]),
              THDVL1=float(p[5]),
              T,
              MachineName = str(p[7]),
              HUM = int(p[8]),
              OLI = int(p[9]),
              GridStatus = bool(p[10])
           )
      ).toDF()

Здесь вам не хватает значения для аргумента T (make it T = p[6]), т.е. вы получаете исключение.

В Python, когда передается любое количество аргументов, обязательно, чтобы вам сначала нужно было передать позиционный аргумент, а затем ключевые аргументы.

Здесь вы используете Row Class, поэтому каждый аргумент должен быть либо ключевым, либо позиционным.

Если использовать оба, то позиционное должно быть первым. как

iot_f = iot_f.map(
          lambda p: Row(
              str(p[0]), 
              p[1], 
              float(p[2]),
              Sum_PF=float(p[3]),
              FRQ=int(p[4]),
              THDVL1=float(p[5]),
              T = p[6],
              MachineName = str(p[7]),
              HUM = int(p[8]),
              OLI = int(p[9]),
              GridStatus = bool(p[10])
           )
      ).toDF()

Answer 2

Я обнаружил ошибку.

iot_f = iot_f.map(
          lambda p: Row(
              DeviceImei = str(p[0]), 
              DeviceTimeStamp = p[1], 
              Avg_PF=float(p[2]),
              Sum_PF=float(p[3]),
              FRQ=int(p[4]),
              THDVL1=float(p[5]),
              **T = int(p[6])** ,
              MachineName = str(p[7]),
              HUM = int(p[8]),
              OLI = int(p[9]),
              GridStatus = bool(p[10])
           )
      ).toDF()

Теперь он показывает некоторую другую ошибку, которую я выясню отдельно.