spark 例子count(distinct 字段) 例子描述: 有个网站访问日志,有4个字段:(用户id,用户名,访问次数,访问网站) 需要统计: 1.用户的访问总次数去重 2.用户一共访问了多少种不同的网站 这里用sql很好写 select id,name,count(distinct url) from table group by id,name 其实这个题目是继官方和各种地方讲解聚合函数(aggregate)的第二个例子,第一个例子是使用aggregate来求平均数. 我们先用简易版来

Spark小课堂Week7 从Spark中一个例子看面向对象设计 今天我们讨论了个问题,来设计一个Spark中的常用功能. 功能描述:数据源是一切处理的源头,这次要实现下加载数据源的方法load() 初始需求 需求:支持Json数据源加载 具体:输入一个path,需要返回一个Relation, Relation中提供scan()和write()两个方法 示意代码: class Context{ public Relation json(String path){ return new Relat

非交互式运行Spark Application 的例子 $ cat Count.py import sys from pyspark import SparkContext if __name__ == "__main__": sc = SparkContext() logfile = sys.argv[1] count = sc.textFile(logfile).filter(lambda line: '.jpg' in line).count() print "JPG

Spark Python 索引页 [Spark][Python]sortByKey 例子 的继续: [Spark][Python]groupByKey例子 In [29]: mydata003.collect() Out[29]: [[u'00001', u'sku933'], [u'00001', u'sku022'], [u'00001', u'sku912'], [u'00001', u'sku331'], [u'00002', u'sku010'], [u'00003', u'sku88

[Spark][Python]sortByKey 例子: [training@localhost ~]$ hdfs dfs -cat test02.txt00002 sku01000001 sku93300001 sku02200003 sku88800004 sku41100001 sku91200001 sku331[training@localhost ~]$ mydata001=sc.textFile("test02.txt")mydata002=mydata001.map(l

spark streaming 入门例子: spark shell import org.apache.spark._ import org.apache.spark.streaming._ sc.getConf.setMaster("local").setAppName("RDDTest"); val ssc = new StreamingContext(sc, Seconds(2)); val fileStream = ssc.textFileStream(&q

[学习笔记] windows scala helloworld例子详解: 在操作系统中,我们的Test3.scala会生成Test3.class,然后class文件被虚拟机加载并执行, 这一点和java是一样的. 以object关键字修饰一个类名,这种语法叫做孤立对象,这个对象是单例的. 相当于将单例类和单例对象同时定义.相当于java中的单例,即在内存中只会存在一个Test3实例.创建一个Scala Object,它相当于java的static, 不要用Scala-class去建工程,不然就不

object func { def main(args:Array[String]):Unit={ //函数赋值给变量时, 必须在函数后面加上空格和下划线. def sayHello(name: String) { println("Hello, " + name) } val sayHelloFunc = sayHello _ sayHelloFunc("leo") //Scala定义匿名函数的语法规则: (参数名: 参数类型) => 函数体 val say

object ScalaApp { def main(args: Array[String]): Unit = { var conf = new SparkConf() conf.setMaster("local") var sc = new SparkContext(conf) val lines: RDD[String] = sc.textFile("./words.txt") lines.flatMap(line=>line.split()).reduc

[学习笔记] reduce将RDD中元素前两个传给输入函数,产生一个新的return值,将新产生的return值与RDD中下一个元素(即第三个元素)组成两个元素,再被传给输入函数,这样递归运作,直到最后只有一个值为止.*/    val rdd07 = sc.parallelize(1 to 10)    val sum = rdd07.reduce((x, y) => x + y)    println("sum is " + sum) 文章转载自原文:https://blog

spark用的是cdh spark-2.0.1 package main.scala import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} /* Created by tomtang on 2016/12/16. */ object tomhelloworld { def FILE_NAME:String = "word_count_results_" def main(arg

统计效果: 代码部分: import org.apache.spark.sql.hive.HiveContext import org.apache.spark.{Logging, SparkConf, SparkContext} import org.apache.spark.sql.{DataFrame, Row, SaveMode, _} import com.alibaba.fastjson.{JSON, JSONObject} import org.apache.hadoop.conf

import org.apache.spark._import SparkContext._import java.util.{Calendar,Properties,Date,Locale}import java.text.SimpleDateFormat import java.math.BigDecimal;import java.math.RoundingMode;import java.text.DecimalFormat;import java.text.NumberFormat;i

scala集合转化为DS/DF case class TestPerson(name: String, age: Long, salary: Double) val tom = TestPerson(,35.5) val sam = TestPerson(,40.5) val PersonList = mutable.MutableList[TestPerson]() PersonList += tom PersonList += sam val personDS = PersonList.to

1.利用scala语言开发spark的worcount程序(本地运行) package com.zy.spark import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} //todo:利用scala语言来实现spark的wordcount程序 object WordCount { def main(args: Array[String]): Unit = { //1.创建SparkConf

环境配置 IDE: https://www.jetbrains.com/idea/ 子雨大数据之Spark入门教程(Scala版) /* implement */ 语言特性 Online compiler: https://scastie.scala-lang.org/ 只记录 “不太一样” 的特性. 一.常量变量 常量 val, 变量 var. 二.循环 var name = "hello" for (n <- name) { println(n) } 嵌套循环,遍历数字和数组

1.pom.xml文件配置,直接上代码. <?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apa

 Spark 版本为 3.0.0,默认采用的 Scala 编译版本为 2.12 1. 创建名为spark-project 的项目  2. 将src 删除,把spark-project 当成一个父项目,并在该项目下创建一个moule ,spark-core (子项目)  2.scala-idea 插件下载安装 https://plugins.jetbrains.com/plugin/1347-scala/versions 选择和idea对应的版本的scala插件 3. 添加scala 插件  4.

搞清楚几个概念: 闭包 柯里化 搭建了intellij idea 的scala 开发环境

1. 在生产环境下,很多文件是GBK编码格式的,而SPARK 常用的textFile方法默认是写死了读UTF-8格式的文件,其他格式文件会显示乱码 用如下代码实现读取GBK文件的方法 import org.apache.hadoop.io.{LongWritable, Text}import org.apache.hadoop.mapred.TextInputFormatimport org.apache.spark.rdd.RDDimport org.apache.spark.{SparkCo

1 http://www.cnblogs.com/charlesblc/p/6165201.html [转载]协同过滤 & Spark机器学习实战 2 基于Spark构建推荐引擎之一:基于物品的协同过滤推荐: http://blog.csdn.net/sunbow0/article/details/42737541 3 2017年,你还在用用户画像和协同过滤做推荐系统吗? http://www.infoq.com/cn/articles/user-portrait-collaborative-f