本文总结在使用ambari时产生的异常，以及如何解决的。
如果发生了异常，在界面上不能直观的看出异常的原因，那么我一般通过查看日志的方法解决。
通过下面的命令查看1
vim /var/log/ambari-server/ambari-server.log
1 | Error executing bootstrap Cannot create /var/run/ambari-server/bootstrap |
记录spark-submit提交Spark程序出现的一个异常，以供第一次出现这种异常且不知道原因，该怎么解决的的同学参考。
1 | Exception in thread "main" org.apache.spark.SparkException: Application application_1529650293575_0148 finished with failed status |
本文总结如何将DataFrame按某列降序排序，因为Spark默认的排序方式为升序，而降序的用法和java语言等又不一样，所以需要特地总结记录一下其用法。
1 | val data = Array((7, 2, 3), (1, 8, 6), (4, 5, 9)) |
1 | +----+----+----+ |
1 | df.orderBy("col2").show() |
1 | +----+----+----+ |
本文讲解Spark如何获取当前分区的partitionId，这是一位群友提出的问题，其实只要通过TaskContext.get.partitionId（我是在官网上看到的），下面给出一些示例。
下面的代码主要测试SparkSession，SparkContext创建的rdd和df是否都支持。1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37package com.dkl.leanring.partition
import org.apache.spark.sql.SparkSession
import org.apache.spark.TaskContext
/**
* 获取当前分区的partitionId
*/
object GetPartitionIdDemo {
def main(args: Array[String]): Unit = {
val spark = SparkSession.builder().appName("GetPartitionIdDemo").master("local").getOrCreate()
val sc = spark.sparkContext
val data = Seq(1, 2, 3, 4)
// 测试rdd,三个分区
val rdd = sc.parallelize(data, 3)
rdd.foreach(i => {
println("partitionId：" + TaskContext.get.partitionId)
})
import spark.implicits._
// 测试df,三个分区
val df = rdd.toDF("id")
df.show
df.foreach(row => {
println("partitionId：" + TaskContext.get.partitionId)
})
// 测试df,两个分区
val data1 = Array((1, 2), (3, 4))
val df1 = spark.createDataFrame(data1).repartition(2)
df1.show()
df1.foreach(row => {
println("partitionId：" + TaskContext.get.partitionId)
})
}
}
本文利用SparkStreaming+Kafka实现实时的统计uv，即独立访客，一个用户一天内访问多次算一次，这个看起来要对用户去重，其实只要按照WordCount的思路，最后输出key的数量即可，所以可以利用SparkStreaming+Kafka 实现基于缓存的实时wordcount程序，这里稍加改动，如果uv数量增加的话就打印uv的数量(key的数量)。
数据是我随机在kafka里生产的几条，用户以空格区分开（因为用的之前单词统计的程序）
首先在kafka建一个程序用到topic:KafkaUV1
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic KafkaUV
我的目录为：/spark/dkl/kafka/UV_checkpoint1
hadoop fs -mkdir -p /spark/dkl/kafka/UV_checkpoint
tag:
缺失模块。
1、请确保node版本大于6.2
2、在博客根目录（注意不是yilia根目录）执行以下命令：
npm i hexo-generator-json-content --save
3、在根目录_config.yml里添加配置：
jsonContent:
meta: false
pages: false
posts:
title: true
date: true
path: true
text: false
raw: false
content: false
slug: false
updated: false
comments: false
link: false
permalink: false
excerpt: false
categories: false
tags: true