使用 PySpark 在 HDFS 中保存并附加文件

我在 PySpark 中有一个名为df。我已经注册了这个df as a temptable像下面这样。

df.registerTempTable('mytempTable')

date=datetime.now().strftime('%Y-%m-%d %H:%M:%S')

现在从这个临时表中我将获得某些值，例如列的 max_idid

min_id = sqlContext.sql("select nvl(min(id),0) as minval from mytempTable").collect()[0].asDict()['minval']

max_id = sqlContext.sql("select nvl(max(id),0) as maxval from mytempTable").collect()[0].asDict()['maxval']

现在我将收集所有这些值，如下所示。

test = ("{},{},{}".format(date,min_id,max_id))

我找到test不是一个data frame但这是一个str string

>>> type(test)
<type 'str'>

现在我想保存这个test作为文件HDFS。我还想将数据附加到同一文件中hdfs.

我如何使用 PySpark 做到这一点？

仅供参考，我正在使用 Spark 1.6，无法访问 Databricksspark-csv包裹。

在这里，您只需要将数据与concat_ws并将其修改为文本：

query = """select concat_ws(',', date, nvl(min(id), 0), nvl(max(id), 0))
from mytempTable"""

sqlContext.sql(query).write("text").mode("append").save("/tmp/fooo")

或者甚至是更好的选择：

from pyspark.sql import functions as f

(sqlContext
    .table("myTempTable")
    .select(f.concat_ws(",", f.first(f.lit(date)), f.min("id"), f.max("id")))
    .coalesce(1)
    .write.format("text").mode("append").save("/tmp/fooo"))