小编phi*_*ert的帖子

我正在运行为Hadoop-2.4预先构建的Spark-1.4.0(在本地模式下)来计算DoubleRDD的平方和.我的Scala代码看起来像

sc.parallelize(Array(2., 3.)).fold(0.0)((p, v) => p+v*v)

它给出了令人惊讶的结果97.0.

与Scala版本相比,这是非常反直觉的 fold

Array(2., 3.).fold(0.0)((p, v) => p+v*v)

这给出了预期的答案13.0.

由于缺乏理解,我很可能在代码中犯了一些棘手的错误.我已经读过有关如何使用的函数RDD.fold()应该是通信的,否则结果可能取决于分区等.例如,如果我将分区数更改为1,

sc.parallelize(Array(2., 3.), 1).fold(0.0)((p, v) => p+v*v)

代码会169.0在我的机器上给我!

有人可以解释这里究竟发生了什么吗？

我有一个包含 JSON 数据的文本文件，如下所示：

{
  "element" : value,
  "id" : value,
  "total" : []
}
{
  "element" : value,
  "id" : value,
  "total: []
}

所有 JSON 都以新行分隔。

我正在尝试将所有文​​本文件数据加载到临时视图中：

sqlContext.read.textFiles("/path").createOrReplaceTempView("result")

val data = sqlContext.sql("select * from result").collect()

结果：

[{"element" : value,"id" : value,"total" : [] }]
[{"element" : value,"id" : value, "total" : []}]

我需要提取 id 和与之相关的总数。

有没有办法在火花中处理这个？

我有一个 CSV 文件，如下所示：

"a","b","c","{""x"":""xx"",""y"":""yy""}"

当我使用 java CSV reader ( au.com.bytecode.opencsv.CSVParser) 时，它会在我指示时设法解析字符串defaultEscapeChar = '\u0000'

当我尝试使用 spark 2.2 CSV 阅读器阅读它时，它失败了，无法将其拆分为 4 列。这是我尝试过的：

val df = spark.read.format("csv")
              .option("quoteMode","ALL")
              .option("quote", "\u0000")
              .load("s3://...")

我也尝试过，option("escape", "\u0000") 但没有运气。

我需要选择哪些 CSV 选项才能正确解析此文件？

假设我有一个像这样的数据框。

val df = sc.parallelize(Seq(
            (1.0, 1,"Matt"), 
            (1.0, 2,"John"),
            (1.0, 3,null.asInstanceOf[String]),
            (-1.0, 2,"Adam"), 
            (-1.0, 4,"Steve"))
          ).toDF("id", "timestamp","name")

我想获取时间戳排序的每个id的最后一个非空值。这是我的窗户

val partitionWindow = Window.partitionBy($"id").orderBy($"timestamp".desc)

我正在创建一个独特的窗口数据

val filteredDF = df.filter($"name".isNotNull).withColumn("firstName", first("name") over (partitionWindow)).drop("timestamp","name").distinct

并将其重新连接到实际数据

val joinedDF = df.join(filteredDF, windowDF.col("id") === filteredDF.col("id")).drop(filteredDF.col("id"))

joinedDF.show()

它工作正常，但我不喜欢这种解决方案，有人可以建议我做些更好的事情吗？

另外，谁能告诉我为什么最后一个功能不起作用？我试过了，结果不正确

 val partitionWindow = Window.partitionBy($"id").orderBy($"timestamp")

val windowDF = df.withColumn("lastName", last("name") over (partitionWindow))

我有一个包含以下列的表.

C1,C2,Process TimeStamp,InsertDateTimeStamp
p1,v1,2014-01-30 12:15:23,2013-10-01 05:34:23 
p1,v2,2014-01-31 05:11:34,2013-12-01 06:12:31 
p1,v3,2014-01-31 07:16:05,2012-09-01 07:45:20 
p2,v4,2014-02-01 09:22:52,2013-12-01 06:12:31 
p2,v5,2014-02-01 09:22:52,2012-09-01 07:45:20

现在,我想基于最新的每个主键获取唯一行Process TimeStamp.

如果Process TimeStamp相同InsertDateTimeStamp则应选择具有最新的行.

所以,我的结果应该是.

p1,v3,2014-01-31 07:16:05,2012-09-01 07:45:20 
p2,v4,2014-02-01 09:22:52,2013-12-01 06:12:31

如何通过HiveQL实现这一目标？

我目前正在使用hive 0.10.我不能将子查询与IN或EXISTS一起使用.

谢谢.

• 表描述信息

hive> desc log23;
OK
col_name        data_type       comment
17/05/25 10:49:12 INFO mapred.FileInputFormat: Total input files to process : 1
host                    string                  from deserializer
remote_host             string                  from deserializer
remote_logname          string                  from deserializer
remote_user             string                  from deserializer
request_time            string                  from deserializer
request_method          string                  from deserializer
request_url             string                  from deserializer
first_line              string                  from deserializer
http_status             string                  from deserializer
bytes                   string                  from deserializer
referer                 string                  from deserializer
agent                   string                  from deserializer
Time taken: 0.049 seconds, Fetched: 12 row(s)

• apache日志格式序列化

serializationLib:org.apache.hadoop.hive.contrib.serde2.RegexSerDe, parameters:{output.format.string=%1$s %2$s %3$s %4$s …

我正在尝试 beeline cli 查询配置单元表并将输出结果存储为变量。使用 beeline 命令：

beeline -u connection_string -n user_name -w password_file \
-e "select count(*) from db.table_name"

使用这个命令，我得到当前的输出结果为：

+---------------+--+ 
| record_count  | 
+---------------+--+ 
| 80785         | 
+---------------+--+

虽然我需要结果为： Record count:80785

我正在使用的另一个命令是：

beeline -u connection_string -n user_name -w password_file \
-e "select * from db.table_name;” > result.csv

这再次以表格格式数据显示结果，以|.

默认情况下，beeline 基本上是返回 header( table_name.column_name)，然后是表格格式的数据。然而，我想消除这一点并获得像 hive CLI 这样的结果。

我正在尝试dropTempTable()在各个临时表的使用结束后使用（以释放内存用于下一次计算）。

较新的 Spark Session 不需要sqlContext，因此，我对如何使用该功能感到困惑。

1）尝试过，我用来注册临时表的同一个DF -

DF.dropTempTable('xyz')

没用。

2) 也尝试了以下方法，因为 spark 在内部sqlContext也与 一起调用sparkContext，但没有用 -

spark.dropTempTable('xyz')

3）试图spark.catalog下降，这也失败了-

spark.catalog.dropTempTable('xyz')

该怎么办？互联网上的 1.6 示例在 2.3 版本中不起作用dropTempTable()。

有什么帮助吗？

val partitionsColumns = "idnum,monthnum"
val partitionsColumnsList = partitionsColumns.split(",").toList
val loc = "/data/omega/published/invoice"
val df = sqlContext.read.parquet(loc)
val windowFunction = Window.partitionBy  (partitionsColumnsList:_*).orderBy(df("effective_date").desc)

<console>:38: error: overloaded method value partitionBy with alternatives:
(cols: org.apache.spark.sql.Column*)     org.apache.spark.sql.expressions.WindowSpec <and>
(colName: String,colNames: String*)org.apache.spark.sql.expressions.WindowSpec
cannot be applied to (String)
val windowFunction = Window.partitionBy(partitionsColumnsList:_*).orderBy(df("effective_date").desc)

Run Code Online (Sandbox Code Playgroud)

是否可以将列列表发送到partitionBy方法 Spark/Scala？

我已经实现了将一列传递给有效的partitionBy方法。我不知道如何将多列传递给partitionByMethod

基本上我想传递List(Columns)给partitionBy方法

Spark 版本是 1.6。

我正在 Databricks 中创建一个增量表，其中包含 1 天的代理日志（数百行数百万行）。我希望能够按小时对表进行分区，因此简单地按“_time”列对表进行分区是不够的。另外，我正在使用%sql运行时在笔记本中创建表，但如果这是更好的选择，我愿意在 Scala 中创建它。

如何创建分区逻辑，以便为每个小时的日志创建一个新分区？

我的专栏的格式_time如下。

yyyy-mm-ddThh:mm:ss.xxxxxx-time:zone其中时间和区域是 2 位数字。

因此，我想对每个独特的小时进行分区，一天有 24 个分区，就像这样。

yyyy-mm-ddThh