小编hi-*_*zir的帖子

在这段注释中的代码中，正确显示了列表缓冲区项的长度，但是在第二注释中，代码从未执行。为什么会发生？

val conf = new SparkConf().setAppName("app").setMaster("local")
val sc = new SparkContext(conf)

var wktReader: WKTReader = new WKTReader(); 
val dataSet = sc.textFile("dataSet.txt")

val items = new ListBuffer[String]() 
dataSet.foreach { e =>
  items += e
  println("len = " + items.length) //1. here length is ok
}

println("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!")
items.foreach { x => print(x)} //2. this code doesn't be executed

日志在这里：

16/11/20 01:16:52 INFO Utils: Successfully started service 'SparkUI' on port 4040.
    16/11/20 01:16:52 INFO SparkUI: Bound SparkUI to 0.0.0.0, and started at http://192.168.56.1:4040 …

我在Jupyter中用类似这样的东西开始我的Spark会话:

from pyspark.sql import SparkSession

spark = ( SparkSession
            .builder
            .appName('analysis')
            .getOrCreate() )

有没有办法找出Web UI使用的端口？谢谢!

我在PySpark中有一个包含空格,Null和Nan的数据帧.我想删除任何有这些行的行.我尝试了以下命令,但似乎没有任何工作.

myDF.na.drop().show()
myDF.na.drop(how='any').show()

以下是数据帧:

+---+----------+----------+-----+-----+
|age|  category|      date|empId| name|
+---+----------+----------+-----+-----+
| 25|electronic|17-01-2018|  101|  abc|
| 24|    sports|16-01-2018|  102|  def|
| 23|electronic|17-01-2018|  103|  hhh|
| 23|electronic|16-01-2018|  104|  yyy|
| 29|       men|12-01-2018|  105| ajay|
| 31|      kids|17-01-2018|  106|vijay|
|   |       Men|       nan|  107|Sumit|
+---+----------+----------+-----+-----+

我错过了什么？解决NULL,Nan或空格的最佳方法是什么,以便在实际计算中没有问题？

我一直在本地集群上运行我的 Spark 作业，该集群具有从中读取输入和写入输出的 hdfs。现在我已经设置了一个 AWS EMR 和一个 S3 存储桶，我可以在其中输入我的输入，并且我希望我的输出也写入 S3。

错误：

用户类抛出异常：java.lang.IllegalArgumentException：错误 FS：s3://something/input，预期：hdfs://ip-some-numbers.eu-west-1.compute.internal:8020

我尝试搜索相同的问题，并且有几个关于此问题的问题。有人建议它仅用于输出，但即使我禁用输出，也会出现相同的错误。

另一个建议是有问题 FileSystem我的代码有问题。以下是我的程序中出现的所有输入/输出：

第一次发生在我的 customFileInputFormat中getSplits(JobContext job)，我实际上并没有修改自己，但我可以：

FileSystem fs = path.getFileSystem(job.getConfiguration());

类似的情况在我的自定义中RecordReader，我自己也没有修改过：

final FileSystem fs = file.getFileSystem(job);

在nextKeyValue()我RecordReader自己写的习惯中，我使用：

FileSystem fs = FileSystem.get(jc);

最后，当我想检测我使用的文件夹中的文件数时：

val fs = FileSystem.get(sc.hadoopConfiguration)
val status = fs.listStatus(new Path(path))

我认为问题出在我的代码上，但是如何修改FileSystem调用以支持来自 S3 的输入/输出？

我正在对 ZFS 进行一些研究，如果它可以用作分布式处理框架的组件。我要回答的主要问题是 - 如果数据驻留在 zfs 中，Apache Spark 是否会以高效的分布式方式运行？

IE。HDFS 上的 Spark 有数据局部性的概念，ZFS 也可以这样说。

如果我有一个名为df的DataFrame看起来像：

+----+----+
|  a1+  a2|
+----+----+
| foo| bar|
| N/A| baz|
|null| etc|
+----+----+

我可以像这样有选择地替换值：

val df2 = df.withColumn("a1", when($"a1" === "N/A", $"a2"))

所以df2看起来像：

+----+----+
|  a1+  a2|
+----+----+
| foo| bar|
| baz| baz|
|null| etc|
+----+----+

但是为什么我不能检查它是否为null，例如：

val df3 = df2.withColumn("a1", when($"a1" === null, $"a2"))

这样我得到：

+----+----+
|  a1+  a2|
+----+----+
| foo| bar|
| baz| baz|
| etc| etc|
+----+----+

编辑：$“ a1” .isNull似乎不起作用。可能是因为我如何构造用于测试的数据框，如下所示？

val schema = StructType(
                StructField("a1", StringType, false) ::
                StructField("a2", StringType, false) :: Nil …

我已经从输入文件加载了一个表。

CREATE TABLE MyTable (
    ID INT,
    VALUE FLOAT,
    RATE  INT
...

LOAD DATA LOCAL INPATH 'MYPATH' INTO TABLE MyTable;

现在我想基于这个创建一个新的

DerivedTable = 
    SELECT ID, VALUE*RATE AS Total
    FROM MyTable
    WHERE VALUE IS NOT NULL;

然后我将使用该表作为其他表和输出的源。

创建此“临时”表的正确 Sql（或 Hive）方法是什么？这应该适用于spark-sql？

PS：我知道如何在 Spark-Shell 中做到这一点。但这不是我要找的。

我想将重命名key下图作为name,_1对rownum,_2对status

  root
  |-- id: string (nullable = true)
  |-- info: map (nullable = true)
  |    |-- key: string
  |    |-- value: struct (valueContainsNull = true)
  |    |    |-- _1: long (nullable = false)
  |    |    |-- _2: string (nullable = true)

请帮忙