2019年 – 第3页

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

bug清单 4月 24,2019

spark错误:scalac: Error: object FloatRef does not have a member create

运行spark项目，出现下面奇怪的错误：

Error:scalac: Error: object FloatRef does not have a member create
scala.reflect.internal.FatalError: object FloatRef does not have a member create

at scala.reflect.internal.Definitions$DefinitionsClass.scala$reflect$internal$Definitions$DefinitionsClass$$fatalMissingSymbol(Definitions.scala:1166)

at scala.reflect.internal.Definitions$DefinitionsClass.getMember(Definitions.scala:1183)

at scala.reflect.internal.Definitions$DefinitionsClass.getMemberMethod(Definitions.scala:1218)

at scala.tools.nsc.transform.LambdaLift$$anonfun$scala$tools$nsc$transform$LambdaLift$$refCreateMethod$1.apply(LambdaLift.scala:41)

at scala.tools.nsc.transform.LambdaLift$$anonfun$scala$tools$nsc$transform$LambdaLift$$refCreateMethod$1.apply(LambdaLift.scala:41)

at scala.reflect.internal.util.Collections$$anonfun$mapFrom$1.apply(Collections.scala:182)

at scala.reflect.internal.util.Collections$$anonfun$mapFrom$1.apply(Collections.scala:182)

at scala.collection.immutable.List.map(List.scala:274)

at scala.reflect.internal.util.Collections$class.mapFrom(Collections.scala:182)

at scala.reflect.internal.SymbolTable.mapFrom(SymbolTable.scala:16)

at scala.tools.nsc.transform.LambdaLift.scala$tools$nsc$transform$LambdaLift$$refCreateMethod$lzycompute(LambdaLift.scala:41)

这是scala依赖库的问题，把
最后把IDEA依赖的scala版本改成scala2.10.4 就可以。

在IDEA下载scala很慢，给相关scala2.10.4下载地址:

https://scala-lang.org/files/archive/scala-2.10.4.zip

作者 east

bug清单 4月 22,2019

Spark Uri错误：java.lang.IllegalArgumentException: Illegal character in opaque part at index 5

val destinationPath = "file:\\E:\\newcode\\MyFirstProject\\data\\stockresult.txt" ;
FileSystem fs = FileSystem.get(URI.create(path),conf);
writer = new BufferedWriter(new OutputStreamWriter(fs.create(new Path(path))));
if(null!=writer){
    logger.info("[HdfsOperate]>> initialize writer succeed!");
}

出现了下面的错误：

Exception in thread “main” java.lang.IllegalArgumentException: Illegal character in opaque part at index 5: file:\E:\newcode\MyFirstProject\data\stockresult.txt
at java.net.URI.create(URI.java:852)

问题是Uri格式没写对，写成下面这样就对了

val destinationPath = "file:/E:/newcode/MyFirstProject/data/stockresult.txt" ;

作者 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

bug清单 4月 19,2019

spark错误：Unable to find encoder for type stored in a Dataset

运行下面代码，出现了错误

import spark.implicits._
val tmpSeq = JavaConverters.asScalaIteratorConverter(list.iterator).asScala.toSeq
val ds = spark.createDataset(tmpSeq)

Error:(32, 33) Unable to find encoder for type stored in a Dataset. Primitive types (Int, String, etc) and Product types (case classes) are supported by importing spark.implicits._ Support for serializing other types will be added in future releases.

觉得因为Dataset是强类型，没有指定好类型，于是修改如下面这样，果然错误消失了：

 import spark.implicits._
 // List 转 Seq
 val tmpSeq = JavaConverters.asScalaIteratorConverter(list.iterator).asScala.toSeq
 sinaStockStream.close()
// spark.sparkContext.parallelize(tmpSeq);
val personEncoder: Encoder[KLineModel] = Encoders.bean(classOf[KLineModel])
 val ds = spark.createDataset(tmpSeq)(personEncoder)

作者 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

年度归档2019