大数据开发 – 第38页

Spark 5月 7,2019

解决Spark读取CSV文件中文乱码的完整例子

park.read.option(“header”,”true”).csv(path) 的默认方法，如果读取的源数据是utf-8k中文的，能正常显示，但如果Spark读取带有GBK或GB2312等中文编码的话，就会有Spark GBK乱码或Spark GB2312乱码。下面示例一个完整例子，如果Spark读取不是GBK编码的，只需要替换下面的中文编码。

import java.sql.DriverManager

import org.apache.hadoop.fs.Path
import org.apache.hadoop.io.{LongWritable, Text}
import org.apache.hadoop.mapred.TextInputFormat
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.types.{StructField, StructType, _}

import scala.collection.mutable.ArrayBuffer

/**
* 通过友盟统计1天访问明细
*/
object UmengRangfeiSQL {
def main(arg: Array[String]): Unit = {
val spark = SparkSession.builder().appName(“UmengRangfeiSQL”).master(“local[*]”).getOrCreate(); //为读取的数据创建schema
// println(System.getProperty(“file.encoding”))
// val pps = System.getProperties
// pps.setProperty(“file.encoding”, “GB2312”)
val taxiSchema = StructType(Array(
StructField(“PageUrl”, StringType, true),
StructField(“PV”, IntegerType, true),
StructField(“UV”, IntegerType, true),
StructField(“IP”, IntegerType, true),
StructField(“PageViews”, DoubleType, true),
StructField(“Output PV”, IntegerType, true),
StructField(“Stay Time”, StringType, true)
))
val path = “E:\\newcode\\MyFirstProject\\data\\rangfei”
//跳过第一行的标题 .option(“header”,”true”)
// val data = spark.read.option(“header”,”true”).schema(taxiSchema).csv(path)

val mySchema = new ArrayBuffer[String]();
mySchema.append(“PageUrl”);
mySchema.append(“PV”);
mySchema.append(“UV”);
mySchema.append(“IP”);
mySchema.append(“PageViews”);
mySchema.append(“Output PV”);
mySchema.append(“Stay Time”);

val data = readCSV(spark, “TRUE”, mySchema,”GBK”, path)
data.show()

data.createTempView(“umng_rangfei”)
val df = data.toDF()

df.persist()

//按受欢迎的分类倒序排列
val resultRdd = df.sqlContext.sql(“select * from umng_rangfei order by PageViews DESC”)
resultRdd.show()

//过虑查找深度好文
val haowenRdd = df.sqlContext.sql(“select * from umng_rangfei WHERE PageUrl LIKE ‘%haowen%’ AND PV > 100 order by PageUrl DESC”)
haowenRdd.show()

spark.sparkContext.hadoopConfiguration.setBoolean(“mapreduce.input.fileinputformat.input.dir.recursive”, true)
deleteOutPutPath(spark.sparkContext,”E:\\newcode\\MyFirstProject\\data\\output\\haowen”)

//加上repartition来控制只有1个输出文件
haowenRdd.repartition(1).write.format(“com.databricks.spark.csv”).save(“E:\\newcode\\MyFirstProject\\data\\output\\haowen”)

val womanRdd = df.sqlContext.sql(“select * from umng_rangfei WHERE PageUrl LIKE ‘%/woman/?p=%’ AND PV > 100 order by PageUrl DESC”)
womanRdd.show()

deleteOutPutPath(spark.sparkContext,”E:\\newcode\\MyFirstProject\\data\\output\\woman”)

//加上repartition来控制只有1个输出文件
womanRdd.repartition(1).write.format(“com.databricks.spark.csv”).save(“E:\\newcode\\MyFirstProject\\data\\output\\woman”)

}

/**
* 删除文件夹或文件
* @param sc
* @param outputPath
*/
def deleteOutPutPath(sc: SparkContext,outputPath: String):Unit={
val path = new Path(outputPath)
val hadoopConf = sc.hadoopConfiguration
val hdfs = org.apache.hadoop.fs.FileSystem.get(hadoopConf)
if(hdfs.exists(path)){
hdfs.delete(path,true)
}
}

def readCSV(spark:SparkSession,headerSchema:String,mySchema: ArrayBuffer[String],code:String,file:String) ={
val rddArr:RDD[Array[String]] = spark.sparkContext.hadoopFile(file, classOf[TextInputFormat],
classOf[LongWritable], classOf[Text]).map(
pair => new String(pair._2.getBytes, 0, pair._2.getLength, code))
//处理同一个单元格同时出现引号逗号串列问题切割
.map(_.trim.split(“,(?=([^\”]*\”[^\”]*\”)*[^\”]*$)”,-1))
val fieldArr = rddArr.first()
//Row.fromSeq(_) 如果只是 map(Row(_)),会导致 spark.createDataFrame(rddRow,schema)错误
val rddRow = rddArr.filter(!_.reduce(_+_).equals(fieldArr.reduce(_+_))).map(Row.fromSeq(_))
val schemaList = ArrayBuffer[StructField]()
if(“TRUE”.equals(headerSchema)){
for(i <- 0 until fieldArr.length){
println(“fieldArr(i)=” + fieldArr(i))
schemaList.append(StructField(mySchema(i),DataTypes.StringType))
}
}else{
for(i <- 0 until fieldArr.length){
schemaList.append(StructField(s”_c$i”,DataTypes.StringType))
println(“fieldArr(i)=” + fieldArr(i))
}
}
val schema = StructType(schemaList)
spark.createDataFrame(rddRow,schema)
}

}

作者 east

Spark 5月 7,2019

解决Spark读取文本中文乱码的完整例子

spark.read.textFile() 的默认方法，如果读取的源数据是utf-8k中文的，能正常显示，但如果带有GBK或GB2312等中文编码的话，就会有乱码。下面示例一个完整例子，如果不是GBK编码的，只需要替换下面的中文编码。



//导入隐饰操作，否则RDD无法调用toDF方法


object ExcelStockEarn extends AppConf {

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

  def readExcel   = {


    import spark.implicits._


    import org.apache.hadoop.io.{LongWritable, Text}
    import org.apache.hadoop.mapred.TextInputFormat
    val path = "E:\\newcode\\MyFirstProject\\data\\stockearn"
    val inputRdd = spark.sparkContext.hadoopFile(path, classOf[TextInputFormat],
      classOf[LongWritable], classOf[Text]).map(
      pair => new String(pair._2.getBytes, 0, pair._2.getLength, "GBK"))


        //我们要统计脏数据的数量，所以我在这里定义了一个计数器
        val accum = spark.sparkContext.longAccumulator("Error Accum")

        val listRdd =inputRdd
          .map({
            line =>
              val fields = line.split("\\s+")
              if(fields.length == 14) {
                CaseFlow(fields(0).toString, fields(1).toString, fields(2).toString, fields(3).toInt, fields(4).toDouble, fields(5).toString, fields(6).toDouble, fields(7).toDouble,
                  fields(8).toString, fields(9).toString, fields(10).toString, fields(11).toString, fields(12).toString, "")
              }else{
                accum.add(1L)
                CaseFlow(fields(0).toString, "", "", 0, 0, "", 0, 0, "", "", "", "", "", "")
              }

          })
    val  dataRdd = listRdd.filter(_.stockCode.length > 1).toDF()

    dataRdd.createTempView("option_stock")
    val df = dataRdd.toDF()
    df.persist()

    val resultRdd = df.sqlContext.sql("select * from option_stock ")
    resultRdd.show();

    val groupRdd = df.sqlContext.sql("select stockCode, SUM(dealAmount) from option_stock group by StockCode order by StockCode")
    groupRdd.show();

  }




  case class CaseFlow(dealDate : String ,stockCode : String , stockName: String , dealNum : Int , dealPrice : Double, dealContent : String,
                      dealAmount :Double, remainAmount: Double, standby1: String, standby2: String, standby3: String, standby4: String, standby5: String,standby6: String);

}

作者 east

Spark 5月 6,2019

Spark读取文件来统计股票资金流水

数据格式：

操作代码：

import org.apache.hadoop.io.{LongWritable, Text}
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat
import org.apache.spark.sql.types._

import scala.io.Source
//导入隐饰操作，否则RDD无法调用toDF方法


object ExcelStockEarn extends AppConf {

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

  def readExcel   = {


    import spark.implicits._
    val path = "E:\\newcode\\MyFirstProject\\data\\stockearn"

        //我们要统计脏数据的数量，所以我在这里定义了一个计数器
        val accum = spark.sparkContext.longAccumulator("Error Accum")

        val listRdd = spark.read.textFile(path).rdd
          .map({
            line =>
              val fields = line.split("\\s+")
              if(fields.length == 14) {
                CaseFlow(fields(0).toString, fields(1).toString, fields(2).toString, fields(3).toInt, fields(4).toDouble, fields(5).toString, fields(6).toDouble, fields(7).toDouble,
                  fields(8).toString, fields(9).toString, fields(10).toString, fields(11).toString, fields(12).toString, "")
              }else{
                accum.add(1L)
                CaseFlow(fields(0).toString, "", "", 0, 0, "", 0, 0, "", "", "", "", "", "")
              }

          }).toDF();


    listRdd.createTempView("option_stock")
    val df = listRdd.toDF()
    df.persist()

    val resultRdd = df.sqlContext.sql("select * from option_stock ")
    resultRdd.show();

    val groupRdd = df.sqlContext.sql("select stockCode, SUM(dealAmount) from option_stock group by StockCode order by StockCode")
    groupRdd.show();

  }


  case class CaseFlow(dealDate : String ,stockCode : String , stockName: String , dealNum : Int , dealPrice : Double, dealContent : String,
                      dealAmount :Double, remainAmount: Double, standby1: String, standby2: String, standby3: String, standby4: String, standby5: String,standby6: String);

}

作者 east

Spark 4月 25,2019

Spark清洗数据实例-评分数据

 
数据清洗， 是大数据分析过程中重要的环节，主要作用是去除不需要的数据，填充缺失内容，确定缺失值的范围并制定好应对策略。


/**
  * 接收用户的评分信息 
  */
case class UserRating(userId:Int, movieId:Int, rating:Double)


import com.zxl.caseclass.{UserRating, Users}
import com.zxl.conf.AppConf
import com.zxl.datacleaner.UserETL._
import org.apache.spark.sql.SaveMode

/** 
  * 数据格式如下
  * 1,1193,5,978300760
  * 1,661,3,978302109
  * 1,914,3,978301968
  * 1,3408,4,978300275
  * 1,2355,5,978824291
  * 1,1197,3,978302268
  * 1,1287,5,978302039
  * 1,2804,5,978300719
  * 1,594,4,978302268
  * 1,919,4,978301368
  * 1,595,5,978824268
  * 1,938,4,978301752
  * 1,2398,4,978302281
  * 1,2918,4,978302124
  * 1,1035,5,978301753
  * 1,2791,4,978302188
  * 1,2687,3,978824268
  * 1,2018,4,978301777
  * 1,3105,5,978301713
  * 1,2797,4,978302039
  */
object RatingETL extends AppConf {

  def main(args: Array[String]) {

    import sqlContext.implicits._

    // 2 读取样本数据
   // val data_path = "hdfs://movie1:9000/movie/data/ratings.txt"
   val data_path = "data/ratings.dat"
    val data = sc.textFile(data_path, 8)
    val userdata = data.map(_.split(",")).map(f => UserRating(f(0).toInt,f(1).toInt,f(2).toDouble)).cache()

    val userDF = userdata.toDF()
    // 存储结果至数据库
    userDF.write.mode(SaveMode.Append).jdbc(jdbcURL, ratingTable, prop)
  }
}

作者 east

Spark 4月 19,2019

Spark分析个股的活跃性

在前文scala获取免费的股票日k线数据，本文做进一步扩展，统计一下股票振幅，统计最近20天振幅大于4个点有多少天。对于炒股喜欢短线的人来说，振幅大的股票，越好做T+0操作。


import java.util
import java.util.Collections

import org.apache.spark.SparkContext
import org.apache.spark.sql.{Encoder, Encoders, SparkSession}
import stock.SinaStock

import scala.io.Source
object KLineAnalyse {
  def main(args: Array[String]): Unit = {
    println("查询日k线股票 http://data.gtimg.cn/flashdata/hushen/daily/19/sh603000.js")
    val sinaStockStream = Source.fromURL("http://data.gtimg.cn/flashdata/hushen/daily/19/sh603000.js","utf-8")
    val sinaLines=sinaStockStream.getLines
    val spark = SparkSession.builder().appName("kline").master("local[*]").getOrCreate(); //为读取的数据创建schema
    //val sc = new SparkContext(SparkUtils.getSparkConf("SequenceFileUsage"))
    val list = new util.ArrayList[KLineModel]()
    for(line <- sinaLines) { /** 将每行数据解析成SinaStock对象，并答应对应的股票信息 **/
      if(line.length > 20) {
       // println(new KLineModel(line).toString)
        list.add(new KLineModel(line));
      }
      }
    Collections.reverse(list);
    import scala.collection.JavaConverters
    import scala.collection.Seq
    import spark.implicits._
    // List 转 Seq
    val tmpSeq = JavaConverters.asScalaIteratorConverter(list.iterator).asScala.toSeq
    sinaStockStream.close()
    val mySparkRdd = spark.sparkContext.parallelize(tmpSeq);
    val top2Rdd = mySparkRdd.take(20)
    val rateRdd = top2Rdd.map(a => (a.dateStr, (a.highPrice - a.lowPrice)/a.openPrice * 100))
    //过滤每天振幅大于几个点的
    val resultRdd = rateRdd.filter(_._2 >= 4)
    println(resultRdd.toBuffer)
    val wordPairs = resultRdd.map(word => (word, 1))
    val wordCounts = wordPairs.length / 20.0f;
    println("wordCounts: " + wordCounts)

    }

}

作者 east

Spark 4月 18,2019

scala获取免费的股票日k线数据

接口的的抓取使用了Scala标准库的Source


class KLineModel {
  var dateStr ="";
  var openPrice = 0f;
  var closePrice = 0f;
  var highPrice = 0f;
  var lowPrice = 0f;

  private var stockInfo :String =""

  def this(stockInfo:String)
  {
    this()
    this.stockInfo=stockInfo /** 根据腾讯的数据接口解析数据 **/
  val stockDetail=stockInfo.split(Array(' ',' ',' ',' ',' '))
    if (stockDetail.length>4){
      this.dateStr=stockDetail(0)
      this.openPrice=stockDetail(1).toFloat
      this.closePrice =stockDetail(2).toFloat
      this.highPrice=stockDetail(3).toFloat
      this.lowPrice =stockDetail(4).toFloat

    }
  }


  override def toString = s"KLineModel($dateStr, $openPrice, $closePrice, $highPrice, $lowPrice)"

import scala.io.Source
object KLineAnalyse {
  def main(args: Array[String]): Unit = {
    println("查询日k线股票 http://data.gtimg.cn/flashdata/hushen/daily/19/sh603000.js")
    val sinaStockStream = Source.fromURL("http://data.gtimg.cn/flashdata/hushen/daily/19/sh603000.js","utf-8")
    val sinaLines=sinaStockStream.getLines
    for(line <- sinaLines) { /** 将每行数据解析成SinaStock对象，并答应对应的股票信息 **/
      if(line.length > 20) {
        println(new KLineModel(line).toString)
      }
      }
      sinaStockStream.close()
      }

}

作者 east

Spark 4月 18,2019

spark中删除文件夹或文件

这个方法能删除HDFS或本地的文件夹或文件，

val spark = SparkSession.builder().appName("USQL").master("local[*]").getOrCreate(); 
deleteOutPutPath(spark.sparkContext,"E:\\newcode\\MyFirstProject\\data\\output\\")

/**
  * 删除文件夹或文件
  * @param sc
  * @param outputPath
  */
def deleteOutPutPath(sc: SparkContext,outputPath: String):Unit={
  val path = new Path(outputPath)
  val hadoopConf = sc.hadoopConfiguration
  val hdfs = org.apache.hadoop.fs.FileSystem.get(hadoopConf)
  if(hdfs.exists(path)){
    hdfs.delete(path,true)
  }
}

如果是删除文件夹的，前面要加下面的话


spark.sparkContext.hadoopConfiguration.setBoolean("mapreduce.input.fileinputformat.input.dir.recursive", true)

作者 east

Spark 2月 13,2019

Spark本地向量(Local vector)详细解析

一个本地向量拥有从0开始的integer类型的索引以及double类型的值，它保存在单台机器上面。MLlib支持两种类型的本地向量：稠密(dense)向量和稀疏(sparse)向量。一个稠密向量通过一个double类型的数组保存数据，这个数组表示向量的条目值(entry values)；一个稀疏向量通过两个并行的数组（indices和values）保存数据。例如：

向量（5.2，0.0，5.5）

密集向量表示：[5.2，0.0，5.5]

稀疏向量表示：（3，[0,2]，[5.2,5.5]） # 3是向量（5.2，0.0，5.5）的长度，除去0值外，其他两个值的索引和值分别构成了数组[0,2]和数组[5.2,5.5]。

Vector是所有局部向量的基类，Dense-Vector和SparseVector都是Vector的具体实现。

本地向量的基类是Vector，Spark提供了两种实现： DenseVector和SparseVector。 Spark官方推荐使用Vectors中实现的工厂方法去创建本地向量。下面是创建本地向量的例子。

 
 
import org.apache.spark.mllib.linalg.{Vector, Vectors} 
// 创建一个dense vector (5.2, 0.0, 5.5).
 val dv: Vector = Vectors.dense(5.2, 0.0, 5.5) 
// 创建一个sparse vector (5.2, 0.0, 5.5)并且指定它的索引和值
 val sv1: Vector = Vectors.sparse(3, Array(0, 2), Array(5.2, 5.5)) 
// 创建一个sparse vector (5.2, 0.0, 5.5)并且指定它的索引和值，通过指定非0的值，位置0是5.2，位置2是5.5 
val sv2: Vector = Vectors.sparse(3, Seq((0, 5.2), (2, 5.5)))

注意，Scala默认引入scala.collection.immutable.Vector，这里我们需要主动引入MLLib中的org.apache.spark.mllib.linalg.Vector来操作。我们可以看看Vectors对象的部分方法。

作者 east

Hadoop 2月 12,2019

yarn权限问题

① YARN 执行作业遇到 Unauthorized request to start container 问题：

datanode 与 namenode 之间未设置时间同步，所以引起该异常。解决方案：多个 datanode 与 namenode 进行时间同步。

②HDFS 客户端的权限错误：Permission denied

1、在系统的环境变量或 Java JVM 变量里面添加HADOOP_USER_NAME，这个值具体等于多少看自己的情况，以后会运行 Hadoop 上的 Linux 的用户名。（修改完重启 eclipse，不然可能不生效）

2、将当前系统的帐号修改为 hadoop

3、使用 HDFS 的命令行接口修改相应目录的权限，hadoop fs -chmod 777 /user,后面的/user 是要上传文件的路径，不同的情况可能不一样，比如要上传的文件路径为hdfs://namenode/user/xxx.doc，则这样的修改可以，如果要上传的文件路径为hdfs://namenode/Java/xxx.doc，则要修改的为 hadoop fs -chmod 777 /java 或者 hadoop fs – chmod 777 /，java 的那个需要先在 HDFS 里面建立 Java 目录，后面的这个是为根目录调整权限。

推荐第一种

作者 east

Hadoop 2月 12,2019

hadoop环境变量问题

bash:haoop:command not found

今天配置完 Hadoop 运行 hadoop 命令 hadoop fs -ls 时出现：

bash:haoop:command not found

后来经过检查，原来是配置文件中 HADOOP_HOME 变量引用前少了个$，修改完毕后可以正常运行了。

vi /etc/profile

export JAVA_HOME=/usr/Java/jdk1.7.0_75 export HADOOP_HOME=/opt/hadoop-2.6.0

export PATH=.:$HADOOP_HOME/bin:$JAVA_HOME/bin:$PATH

修改完后运行 source /etc/profile 让其立即生效！

在 start-all.sh 是出现异常：

Error:Cannot find configuration directory:/etc/hadoop

解决办法：

在 hadoop-env.sh 配置文件中将

export HADOOP_CONF_DIR=${HADOOP_CONF_DIR:-“etc/hadoop”}

替换成

export HADOOP_CONF_DIR=/usr/hadoop-2.6.0/etc/hadoop

修改完后运行 source hadoop-env.sh 让其立即生效！

作者 east

Hadoop 2月 12,2019

namenode常见问题

1、cannot delete name node is in safe mode

问题：向 hdfs put 数据的时候，导致了 name node is in safe mode，然后使

用 Hadoop dfsadmin -safemode leave 后，解除了安全模式。可是再次使用 hdfs put 或 rm

数据，仍旧导致 name node 进入安全模式。

答案：分析了一下，问题是 namenode 所在机器的硬盘满了。因此即使使用了 hadoop dfsadmin -safemode leave 之后，仍旧不能使用 hdfs。

解决办法：

1，删除 namenode 所在机器的一些数据（本地数据） 2，结束安全模式 hadoop dfsadmin -safemode leave 3，可以正常使用 hdfs 了

2、Hadoop namenode 无法启动

最近遇到了一个问题，执行 start-all.sh 的时候发现 JPS 一下 namenode 没有启动。每次开机都得重新格式化一下 namenode 才可以

其实问题就出在tmp 文件，默认的 tmp 文件每次重新开机会被清空，与此同时 namenode

的格式化信息就会丢失

于是我们得重新配置一个 tmp 文件目录

首先在 home 目录下建立一个 hadoop_tmp 目录

sudo mkdir ~/hadoop_tmp

然后修改 Hadoop/conf 目录里面的 core-site.xml 文件，加入以下节点：

<name>hadoop.tmp.dir</name>

<value>/home/chjzh/hadoop_tmp</value>

<description>A base for other temporary directories.</description>

</property>

注意：我的用户是 chjzh 所以目录是/home/chjzh/hadoop_tmp OK 了，重新格式化Namenode

hadoop namenode –format

然后启动 hadoop

start-all.sh

执行下 JPS 命令就可以看到 NameNode 了

作者 east

分类归档大数据开发