小编Rap*_*oth的帖子

我有许多元组(par1,par2),即通过多次重复实验获得的二维参数空间中的点.

我正在寻找计算和可视化置信椭圆的可能性(不确定这是否是正确的术语).这是我在网上找到的一个示例图,用于显示我的意思:

来源:blogspot.ch/2011/07/classification-and-discrimination-with.html

所以原则上我必须将多元正态分布拟合到数据点的二维直方图.有人可以帮我这个吗？

有时(例如,测试和bechmarking)我想强制执行在DataFrame上定义的转换.调用类似动作的AFAIK count并不能确保所有Columns实际计算,show只能计算所有动作的子集Rows(参见下面的示例)

我的解决方案是使用写入DataFrameHDFS df.write.saveAsTable,但这会使我的系统"混乱"我不希望继续使用的表.

那么触发评估的最佳方法是DataFrame什么？

编辑:

请注意,最近还讨论了spark开发者列表:http://apache-spark-developers-list.1001551.n3.nabble.com/Will-count-always-trigger-an-evaluation-of-each-行td21018.html

我做了一个小例子,它表明counton DataFrame不会评估所有内容(使用Spark 1.6.3和spark-master =测试local[2]):

val df = sc.parallelize(Seq(1)).toDF("id")
val myUDF = udf((i:Int) => {throw new RuntimeException;i})

df.withColumn("test",myUDF($"id")).count // runs fine
df.withColumn("test",myUDF($"id")).show() // gives Exception

使用相同的逻辑,这里是一个show不评估所有行的示例:

val df = sc.parallelize(1 to 10).toDF("id")
val myUDF = udf((i:Int) => {if(i==10) throw new RuntimeException;i})

df.withColumn("test",myUDF($"id")).show(5) // runs fine
df.withColumn("test",myUDF($"id")).show(10) // gives Exception

编辑2:对于Eliasah:例外情况说:

org.apache.spark.SparkException: Job …

我正在使用Spark 1.6.1并遇到一个奇怪的行为:我在包含一些输入数据的数据帧上运行一个带有一些繁重计算(物理模拟)的UDF,并构建一个包含许多列的结果-Dataframe(~40 ).

奇怪的是,在这种情况下,我的输入数据帧的每个记录不止一次调用我的UDF(经常是1.6倍),我发现这是不可接受的,因为它非常昂贵.如果我减少列数(例如减少到20),则此行为将消失.

我设法写下一个小脚本,演示了这个:

import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.functions.udf


object Demo {

  case class Result(a: Double)

  def main(args: Array[String]): Unit = {

    val sc = new SparkContext(new SparkConf().setAppName("Demo").setMaster("local[*]"))
    val sqlContext = new SQLContext(sc)
    import sqlContext.implicits._

    val numRuns = sc.accumulator(0) // to count the number of udf calls

    val myUdf = udf((i:Int) => {numRuns.add(1);Result(i.toDouble)})

    val data = sc.parallelize((1 to 100), numSlices = 5).toDF("id")

    // get results of UDF
    var results = data
      .withColumn("tmp", myUdf($"id"))
      .withColumn("result", $"tmp.a")


    // add …

我知道实现接口意味着什么(技术上),但我不确定我是否理解"合同"包含的内容:

比方说,我做的一类MyList,它实现java.util.List(也就是我实现与编译代码的所有方法),是MyList一个List呢？或者我是否需要阅读我覆盖的方法的所有注释,并确保我的实现满足这些行为的"期望"？

我想在DataFrame中添加具有多个值的列的where条件.

例如,它为单值工作.

df.where($"type".==="type1" && $"status"==="completed").

如何为同一列添加多个值,如下所示.

df.where($"type" IN ("type1","type2") && $"status" IN ("completed","inprogress")

我尝试根据DataFrame 在分区列中具有N(假设N=3)不同值的列重新分区DataFrame x,例如:

val myDF = sc.parallelize(Seq(1,1,2,2,3,3)).toDF("x") // create dummy data

我想实现的是repartiton myDF通过x不产生空分区.有没有比这样做更好的方法？

val numParts = myDF.select($"x").distinct().count.toInt
myDF.repartition(numParts,$"x")

(如果我没有指定numParts的repartiton,大多数我的分区是空的(如repartition创建200个分区)...)

由于镶木地板解析空数组的事实,我在写表之前将空数组替换为null.现在,当我阅读表格时,我想做相反的事情:

我有一个具有以下架构的DataFrame:

|-- id: long (nullable = false)
 |-- arr: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- x: double (nullable = true)
 |    |    |-- y: double (nullable = true)

以及以下内容:

+---+-----------+
| id|        arr|
+---+-----------+
|  1|[[1.0,2.0]]|
|  2|       null|
+---+-----------+

我想用空数组替换null数组(id = 2),即

+---+-----------+
| id|        arr|
+---+-----------+
|  1|[[1.0,2.0]]|
|  2|         []|
+---+-----------+

我试过了:

val arrSchema = df.schema(1).dataType

df
.withColumn("arr",when($"arr".isNull,array().cast(arrSchema)).otherwise($"arr"))
.show()

这使 :

java.lang.ClassCastException:org.apache.spark.sql.types.NullType $无法强制转换为org.apache.spark.sql.types.StructType

编辑:我不想"硬编码"我的数组列的任何模式(至少不是结构的模式),因为这可能因情况而异.我只能df在运行时使用架构信息

我顺便使用Spark 2.1,因此我无法使用 …

有人知道如何在不改变绘图命令顺序的情况下改变键内的顺序吗？默认情况下,密钥与各个图的标题具有相同的顺序.

例如:我首先绘制带有fillcu的错误带,然后将平均值绘制为实线.现在我有错误带标题在顶部(在键中),然后是实线的标题.但我想反过来说.

最好的祝愿

为了提高性能(例如,对于连接),建议首先计算表静态.

在Hive我可以做::

analyze table <table name> compute statistics;

在Impala中:

compute stats <table name>;

我的spark应用程序(从hive-tables读取)是否也受益于预先计算的统计数据？如果是,我需要运行哪一个？它们是否都保存了hive Metastore中的统计数据？我在Cloudera 5.5.4上使用spark 1.6.1

注意:在参考文献的火花1.6.1(https://spark.apache.org/docs/1.6.1/sql-programming-guide.html)中,spark.sql.autoBroadcastJoinThreshold我找到了一个提示:

请注意,目前只有运行命令ANALYZE TABLE COMPUTE STATISTICS noscan的Hive Metastore表支持统计信息.

有时,在处理流的步骤之间对流中的每个元素进行"某事"(例如打印)将是方便的,例如用于调试.

一个简单的例子可能看起来像这样,遗憾的是这不会forEach消耗流:

List<String> list = new ArrayList<>();
list.add("one");
list.add("two");
list.add("three");
list.add("four");

List<String> filteredList = 
        list.stream()
        .filter(s -> s.startsWith("t"))
        .forEach(System.out::println)
        .collect(Collectors.toList());

怎么能实现这一目标？