ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 大数据 > Hadoop > Spark2-oneHot编码-标准化-主成分-聚类

Spark2-oneHot编码-标准化-主成分-聚类

原创 Hadoop 作者：长河BI 时间：2016-11-05 16:00:59 0 删除 编辑

  1. // affairs：一年来婚外情的频率
  2. // gender：性别
  3. // age：年龄
  4. // yearsmarried：婚龄
  5. // children：是否有小孩
  6. // religiousness：宗教信仰程度（5分制，1分表示反对，5分表示非常信仰）
  7. // education：学历
  8. // occupation：职业（逆向编号的戈登7种分类）
  9. // rating：对婚姻的自我评分（5分制，1表示非常不幸福，5表示非常幸福）

  11. import org.apache.spark.sql.SparkSession
  12. import org.apache.spark.sql.Dataset
  13. import org.apache.spark.sql.Row
  14. import org.apache.spark.sql.DataFrame
  15. import org.apache.spark.sql.Column
  16. import org.apache.spark.sql.DataFrameReader
  17. import org.apache.spark.rdd.RDD
  18. import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
  19. import org.apache.spark.sql.Encoder
  20. import org.apache.spark.ml.linalg.Vectors
  21. import org.apache.spark.ml.feature.StringIndexer
  22. import org.apache.spark.ml.feature.OneHotEncoder
  23. import org.apache.spark.ml.feature.VectorAssembler
  24. import org.apache.spark.ml.feature.StandardScaler
  25. import org.apache.spark.ml.feature.PCA
  26. import org.apache.spark.ml.clustering.KMeans

  28. scala> val spark = SparkSession.builder().appName("Spark SQL basic example").config("spark.some.config.option", "some-value").getOrCreate()

  30. scala>

  32. scala> // For implicit conversions like converting RDDs to DataFrames
  33. scala> import spark.implicits._


  36. scala> val data: DataFrame = spark.read.format("csv").option("header", true).load("hdfs://ns1/datafile/wangxiao/Affairs.csv")
  37. data: org.apache.spark.sql.DataFrame = [affairs: string, gender: string ... 7 more fields]

  39. scala>

  41. scala> data.cache
  42. res0: data.type = [affairs: string, gender: string ... 7 more fields]

  44. scala>

  46. scala> data.limit(10).show()





  52. scala>

  54. scala> // 转换字符类型，将Double和String的字段分开放

  56. scala> val data1 = data.select(
  57.      | data("affairs").cast("Double"),
  58.      | data("age").cast("Double"),
  59.      | data("yearsmarried").cast("Double"),
  60.      | data("religiousness").cast("Double"),
  61.      | data("education").cast("Double"),
  62.      | data("occupation").cast("Double"),
  63.      | data("rating").cast("Double"),
  64.      | data("gender").cast("String"),
  65.      | data("children").cast("String"))
  66. data1: org.apache.spark.sql.DataFrame = [affairs: double, age: double ... 7 more fields]

  68. scala>

  70. scala> data1.printSchema()
  71. root
  72.  |-- affairs: double (nullable = true)
  73.  |-- age: double (nullable = true)
  74.  |-- yearsmarried: double (nullable = true)
  75.  |-- religiousness: double (nullable = true)
  76.  |-- education: double (nullable = true)
  77.  |-- occupation: double (nullable = true)
  78.  |-- rating: double (nullable = true)
  79.  |-- gender: string (nullable = true)
  80.  |-- children: string (nullable = true)


  83. scala> data1.limit(10).show




  88. scala>

  90. scala> val dataDF = data1
  91. dataDF: org.apache.spark.sql.DataFrame = [affairs: double, age: double ... 7 more fields]

  93. scala>

  95. scala> dataDF.cache()
  96. res4: dataDF.type = [affairs: double, age: double ... 7 more fields]

  98. scala>

  100. scala> //###################################

  102. scala> val indexer = new StringIndexer().setInputCol("gender").setOutputCol("genderIndex").fit(dataDF)
  103. indexer: org.apache.spark.ml.feature.StringIndexerModel = strIdx_19a888aff882

  105. scala> val indexed = indexer.transform(dataDF)
  106. indexed: org.apache.spark.sql.DataFrame = [affairs: double, age: double ... 8 more fields]

  108. scala> // OneHot编码，注意setDropLast设置为false

  110. scala> val encoder = new OneHotEncoder().setInputCol("genderIndex").setOutputCol("genderVec").setDropLast(false)
  111. encoder: org.apache.spark.ml.feature.OneHotEncoder = oneHot_f0f47e0b5b37

  113. scala> val encoded = encoder.transform(indexed)
  114. encoded: org.apache.spark.sql.DataFrame = [affairs: double, age: double ... 9 more fields]

  116. scala> encoded.show()



  120. scala>

  122. scala> val indexer1 = new StringIndexer().setInputCol("children").setOutputCol("childrenIndex").fit(encoded)
  123. indexer1: org.apache.spark.ml.feature.StringIndexerModel = strIdx_7e4d8c69b823

  125. scala> val indexed1 = indexer1.transform(encoded)
  126. indexed1: org.apache.spark.sql.DataFrame = [affairs: double, age: double ... 10 more fields]

  128. scala> val encoder1 = new OneHotEncoder().setInputCol("childrenIndex").setOutputCol("childrenVec").setDropLast(false)
  129. encoder1: org.apache.spark.ml.feature.OneHotEncoder = oneHot_9a8906781325

  131. scala> val encoded1 = encoder1.transform(indexed1)
  132. encoded1: org.apache.spark.sql.DataFrame = [affairs: double, age: double ... 11 more fields]

  134. scala> encoded1.show()



  138. scala>

  140. scala> val encodeDF: DataFrame = encoded1
  141. encodeDF: org.apache.spark.sql.DataFrame = [affairs: double, age: double ... 11 more fields]

  143. scala> encodeDF.show()



  147. scala> encodeDF.printSchema()
  148. root
  149.  |-- affairs: double (nullable = true)
  150.  |-- age: double (nullable = true)
  151.  |-- yearsmarried: double (nullable = true)
  152.  |-- religiousness: double (nullable = true)
  153.  |-- education: double (nullable = true)
  154.  |-- occupation: double (nullable = true)
  155.  |-- rating: double (nullable = true)
  156.  |-- gender: string (nullable = true)
  157.  |-- children: string (nullable = true)
  158.  |-- genderIndex: double (nullable = true)
  159.  |-- genderVec: vector (nullable = true)
  160.  |-- childrenIndex: double (nullable = true)
  161.  |-- childrenVec: vector (nullable = true)


  164. scala>

  166. scala> //#################################

  168. scala> val assembler = new VectorAssembler().setInputCols(Array("affairs", "age", "yearsmarried", "religiousness", "education", "occupation", "rating", "genderVec", "childrenVec")).setOutputCol("features")
  169. assembler: org.apache.spark.ml.feature.VectorAssembler = vecAssembler_8ccd528981cd

  171. scala>

  173. scala> val vecDF: DataFrame = assembler.transform(encodeDF)
  174. vecDF: org.apache.spark.sql.DataFrame = [affairs: double, age: double ... 12 more fields]

  176. scala> vecDF.select("features").show
  177. 16/11/05 15:56:14 WARN Executor: 1 block locks were not released by TID = 11:
  178. [rdd_17_0]
  179. +--------------------+
  180. | features|
  181. +--------------------+
  182. |[0.0,37.0,10.0,3....|
  183. |[0.0,27.0,4.0,4.0...|
  184. |[0.0,32.0,15.0,1....|
  185. |[0.0,57.0,15.0,5....|
  186. |[0.0,22.0,0.75,2....|
  187. |[0.0,32.0,1.5,2.0...|
  188. |[0.0,22.0,0.75,2....|
  189. |[0.0,57.0,15.0,2....|
  190. |[0.0,32.0,15.0,4....|
  191. |[0.0,22.0,1.5,4.0...|
  192. |[0.0,37.0,15.0,2....|
  193. |[0.0,27.0,4.0,4.0...|
  194. |[0.0,47.0,15.0,5....|
  195. |[0.0,22.0,1.5,2.0...|
  196. |[0.0,27.0,4.0,4.0...|
  197. |[0.0,37.0,15.0,1....|
  198. |[0.0,37.0,15.0,2....|
  199. |[0.0,22.0,0.75,3....|
  200. |[0.0,22.0,1.5,2.0...|
  201. |[0.0,27.0,10.0,2....|
  202. +--------------------+
  203. only showing top 20 rows


  206. scala>

  208. scala> // 标准化--均值标准差

  210. scala> val scaler = new StandardScaler().setInputCol("features").setOutputCol("scaledFeatures").setWithStd(true).setWithMean(true)
  211. scaler: org.apache.spark.ml.feature.StandardScaler = stdScal_2e35fbc29084

  213. scala>

  215. scala> // Compute summary statistics by fitting the StandardScaler.

  217. scala> val scalerModel = scaler.fit(vecDF)
  218. scalerModel: org.apache.spark.ml.feature.StandardScalerModel = stdScal_2e35fbc29084

  220. scala>

  222. scala> // Normalize each feature to have unit standard deviation.

  224. scala> val scaledData: DataFrame = scalerModel.transform(vecDF)
  225. scaledData: org.apache.spark.sql.DataFrame = [affairs: double, age: double ... 13 more fields]

  227. scala> // scaledData:DataFrame = [features: vector, scaledFeatures: vector]

  229. scala>

  231. scala> scaledData.select("features", "scaledFeatures").show
  232. 16/11/05 15:56:20 WARN Executor: 1 block locks were not released by TID = 13:
  233. [rdd_17_0]
  234. +--------------------+--------------------+
  235. | features| scaledFeatures|
  236. +--------------------+--------------------+
  237. |[0.0,37.0,10.0,3....|[-0.4413500298573...|
  238. |[0.0,27.0,4.0,4.0...|[-0.4413500298573...|
  239. |[0.0,32.0,15.0,1....|[-0.4413500298573...|
  240. |[0.0,57.0,15.0,5....|[-0.4413500298573...|
  241. |[0.0,22.0,0.75,2....|[-0.4413500298573...|
  242. |[0.0,32.0,1.5,2.0...|[-0.4413500298573...|
  243. |[0.0,22.0,0.75,2....|[-0.4413500298573...|
  244. |[0.0,57.0,15.0,2....|[-0.4413500298573...|
  245. |[0.0,32.0,15.0,4....|[-0.4413500298573...|
  246. |[0.0,22.0,1.5,4.0...|[-0.4413500298573...|
  247. |[0.0,37.0,15.0,2....|[-0.4413500298573...|
  248. |[0.0,27.0,4.0,4.0...|[-0.4413500298573...|
  249. |[0.0,47.0,15.0,5....|[-0.4413500298573...|
  250. |[0.0,22.0,1.5,2.0...|[-0.4413500298573...|
  251. |[0.0,27.0,4.0,4.0...|[-0.4413500298573...|
  252. |[0.0,37.0,15.0,1....|[-0.4413500298573...|
  253. |[0.0,37.0,15.0,2....|[-0.4413500298573...|
  254. |[0.0,22.0,0.75,3....|[-0.4413500298573...|
  255. |[0.0,22.0,1.5,2.0...|[-0.4413500298573...|
  256. |[0.0,27.0,10.0,2....|[-0.4413500298573...|
  257. +--------------------+--------------------+
  258. only showing top 20 rows


  261. scala>

  263. scala> //##########################

  265. scala> // 主成分

  267. scala> val pca = new PCA().setInputCol("scaledFeatures").setOutputCol("pcaFeatures").setK(3).fit(scaledData)
  268. 16/11/05 15:56:21 WARN Executor: 1 block locks were not released by TID = 14:
  269. [rdd_17_0]
  270. 16/11/05 15:56:22 WARN Executor: 1 block locks were not released by TID = 15:
  271. [rdd_17_0]
  272. 16/11/05 15:56:24 WARN BLAS: Failed to load implementation from: com.github.fommil.netlib.NativeSystemBLAS
  273. 16/11/05 15:56:24 WARN BLAS: Failed to load implementation from: com.github.fommil.netlib.NativeRefBLAS
  274. 16/11/05 15:56:25 WARN LAPACK: Failed to load implementation from: com.github.fommil.netlib.NativeSystemLAPACK
  275. 16/11/05 15:56:25 WARN LAPACK: Failed to load implementation from: com.github.fommil.netlib.NativeRefLAPACK
  276. pca: org.apache.spark.ml.feature.PCAModel = pca_8569d580d6e4

  278. scala> pca.explainedVariance.values //解释变量方差
  279. res11: Array[Double] = Array(0.28779526464781313, 0.23798543640278289, 0.11742828783633019)

  281. scala> pca.pc //载荷（观测变量与主成分的相关系数）
  282. res12: org.apache.spark.ml.linalg.DenseMatrix =
  283. -0.12034310848156521 0.05153952289637974 0.6678769450480689
  284. -0.42860623714516627 0.05417889891307473 -0.05592377098140197
  285. -0.44404074412877986 0.1926596811059294 -0.017025575192258197
  286. -0.12233707317255231 0.08053139375662526 -0.5093149296300096
  287. -0.14664751606128462 -0.3872166556211308 -0.03406819489501708
  288. -0.145543746024348 -0.43054860653839705 0.07841454709046872
  289. 0.17703994181974803 -0.12792784984216296 -0.5173229755329072
  290. 0.2459668445061567 0.4915809641798787 0.010477548320795945
  291. -0.2459668445061567 -0.4915809641798787 -0.010477548320795945
  292. -0.44420980045271047 0.240652448514566 -0.089356723885704
  293. 0.4442098004527103 -0.24065244851456588 0.08935672388570405

  295. scala> pca.extractParamMap()
  296. res13: org.apache.spark.ml.param.ParamMap =
  297. {
  298.     pca_8569d580d6e4-inputCol: scaledFeatures,
  299.     pca_8569d580d6e4-k: 3,
  300.     pca_8569d580d6e4-outputCol: pcaFeatures
  301. }

  303. scala> pca.params
  304. res14: Array[org.apache.spark.ml.param.Param[_]] = Array(pca_8569d580d6e4__inputCol, pca_8569d580d6e4__k, pca_8569d580d6e4__outputCol)

  306. scala>

  308. scala> val pcaDF: DataFrame = pca.transform(scaledData)
  309. pcaDF: org.apache.spark.sql.DataFrame = [affairs: double, age: double ... 14 more fields]

  311. scala> // pcaDF:DataFrame = [features: vector, scaledFeatures: vector,pcaFeatures: vector]

  313. scala> pcaDF.cache()
  314. res15: pcaDF.type = [affairs: double, age: double ... 14 more fields]

  316. scala>

  318. scala> pcaDF.printSchema()
  319. root
  320.  |-- affairs: double (nullable = true)
  321.  |-- age: double (nullable = true)
  322.  |-- yearsmarried: double (nullable = true)
  323.  |-- religiousness: double (nullable = true)
  324.  |-- education: double (nullable = true)
  325.  |-- occupation: double (nullable = true)
  326.  |-- rating: double (nullable = true)
  327.  |-- gender: string (nullable = true)
  328.  |-- children: string (nullable = true)
  329.  |-- genderIndex: double (nullable = true)
  330.  |-- genderVec: vector (nullable = true)
  331.  |-- childrenIndex: double (nullable = true)
  332.  |-- childrenVec: vector (nullable = true)
  333.  |-- features: vector (nullable = true)
  334.  |-- scaledFeatures: vector (nullable = true)
  335.  |-- pcaFeatures: vector (nullable = true)


  338. scala> pcaDF.select("features", "scaledFeatures", "pcaFeatures").show
  339. 16/11/05 15:56:36 WARN Executor: 1 block locks were not released by TID = 18:
  340. [rdd_64_0]
  341. +--------------------+--------------------+--------------------+
  342. | features           | scaledFeatures     | pcaFeatures        |
  343. +--------------------+--------------------+--------------------+
  344. |[0.0,37.0,10.0,3....|[-0.4413500298573...|[0.27828160409293...|
  345. |[0.0,27.0,4.0,4.0...|[-0.4413500298573...|[2.42147114101165...|
  346. |[0.0,32.0,15.0,1....|[-0.4413500298573...|[0.18301418047489...|
  347. |[0.0,57.0,15.0,5....|[-0.4413500298573...|[-2.9795960667914...|
  348. |[0.0,22.0,0.75,2....|[-0.4413500298573...|[1.79299133565688...|
  349. |[0.0,32.0,1.5,2.0...|[-0.4413500298573...|[2.65694237441759...|
  350. |[0.0,22.0,0.75,2....|[-0.4413500298573...|[3.48234503794570...|
  351. |[0.0,57.0,15.0,2....|[-0.4413500298573...|[-2.4215838062079...|
  352. |[0.0,32.0,15.0,4....|[-0.4413500298573...|[-0.6964555195741...|
  353. |[0.0,22.0,1.5,4.0...|[-0.4413500298573...|[2.18771069800414...|
  354. |[0.0,37.0,15.0,2....|[-0.4413500298573...|[-2.4259075891377...|
  355. |[0.0,27.0,4.0,4.0...|[-0.4413500298573...|[-0.7743038356008...|
  356. |[0.0,47.0,15.0,5....|[-0.4413500298573...|[-2.6176149267534...|
  357. |[0.0,22.0,1.5,2.0...|[-0.4413500298573...|[2.95788535193022...|
  358. |[0.0,27.0,4.0,4.0...|[-0.4413500298573...|[2.50146472861263...|
  359. |[0.0,37.0,15.0,1....|[-0.4413500298573...|[-0.5123817022008...|
  360. |[0.0,37.0,15.0,2....|[-0.4413500298573...|[-0.9191740114044...|
  361. |[0.0,22.0,0.75,3....|[-0.4413500298573...|[2.97391491782863...|
  362. |[0.0,22.0,1.5,2.0...|[-0.4413500298573...|[3.17940505267806...|
  363. |[0.0,27.0,10.0,2....|[-0.4413500298573...|[0.74585406839527...|
  364. +--------------------+--------------------+--------------------+
  365. only showing top 20 rows


  368. scala>

  370. scala> //#####################################

  372. scala>

  374. scala> // 注意最大迭代次數和轮廓系数

  376. scala> val KSSE = (2 to 10 by 1).par.toList.map { k =>
  377.      | // 聚类
  378.      | // Trains a k-means model.
  379.      | val kmeans = new KMeans().setK(k).setSeed(1L).setFeaturesCol("pcaFeatures")
  380.      | val model = kmeans.fit(pcaDF)
  381.      |
  382.      | // Evaluate clustering by computing Within Set Sum of Squared Errors.
  383.      | val WSSSE = model.computeCost(pcaDF)
  384.      |
  385.      | (k, WSSSE)
  386.      | }
  387. KSSE: List[(Int, Double)] = List((2,2876.20580405469), (3,1680.6647048004902), (4,1395.7184052948346), (5,1239.9362814229812), (6,999.2793106095127), (7,849.0071338527408), (8,737.8560221633246), (9,771.8211752483357), (10,655.7836351785677))

  389. scala>

  391. scala> KSSE.foreach(println)
  392. (2,2876.20580405469)
  393. (3,1680.6647048004902)
  394. (4,1395.7184052948346)
  395. (5,1239.9362814229812)
  396. (6,999.2793106095127)
  397. (7,849.0071338527408)
  398. (8,737.8560221633246)
  399. (9,771.8211752483357)
  400. (10,655.7836351785677)

来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/29070860/viewspace-2127855/ ，如需转载，请注明出处，否则将追究法律责任。

0
0
上一篇： Spark2 DataFrame数据框常用操作
下一篇： Scala zip 拉链操作
请登录后发表评论 登录
全部评论
长河BI

注册时间：2015-06-16

  • 博文量
    15
  • 访问量
    89058

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593