hadoop - 从Pyspark在HDFS中保存文件

我在Hive中有一个空表，我的意思是该表中没有记录。

使用这个空表，我在data frame中创建了一个pyspark

df = sqlContext.table("testing.123_test")

我已经将此data frame注册为临时表

df.registerTempTable('mytempTable')

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

在此表中，我有一列称为id。

现在我想查询临时表，如下所示

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']

现在我想将date，min_id和max_id保存到HDFS中的文件中

我做了如下:

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"))

现在，当我检查HDFS中的文件时，它将显示所有NULL值。

下面是HDFS中输出的文件。

NULL,NULL,NULL

我想要的是

Date,0,0

Here date is the current timestamp
我如何实现自己想要的。

最佳答案

这在scala中，但是您应该可以轻松地将其复制到Python。
您在这里需要的函数是 na.fill 函数。而且，您必须在以下代码中用Python词典替换Scala Maps:

这是您的DF的外观:

scala> nullDF.show
+----+----+----+
|date|   x|   y|
+----+----+----+
|null|null|null|
+----+----+----+

// You have already done this using Python's datetime functions
val format = new java.text.SimpleDateFormat("dd/MM/YYYY HH:mm:ss")
val curr_timestamp = format.format(new java.util.Date())

//Use na fill to replace null values
//Column names as keys in map
//And values are what you want to replace NULL with

val df = nullDF.na.fill(scala.collection.immutable.Map(
         "date" -> ) ,
         "x" -> "0" ,
         "y" -> "0" ) )

这应该给你

+-------------------+---+---+
|               date|  x|  y|
+-------------------+---+---+
|10/06/2017 12:10:20|  0|  0|
+-------------------+---+---+

关于hadoop - 从Pyspark在HDFS中保存文件，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/44468680/