private val DEFAULT_ZOOKEEPER_QUORUM = "127.0.0.1:2181"private lazy val (table, conn) = createConnectiondef bulk(items:Iterator) = {items.foreach(conn.put(_))conn.flush....} ...... }然后保证这个类在map,foreachRDD等函数下使用，譬如：dstream.foreachRDD{ rdd =>rdd.foreachPartition{iter=>SimpleHBaseClient.bulk(iter) } }为什么要保证放到foreachRDD/map 等这些函数里呢？Spark的机制是先将用户的程序作为一个...
kafka与redis的区别： 1、redis消息推送（基于分布式 pub/sub）多用于实时性较高的消息推送，并不保证可靠。其他的mq和kafka保证可靠但有一些延迟（非实时系统没有保证延迟）。redis-pub/sub断电就清空，而使用redis-list作为消息推送虽然有持久化，但是又太弱智，也并非完全可靠不会丢。2、redis 发布订阅除了表示不同的 topic 外，并不支持分组，比如kafka中发布一个东西，多个订阅者可以分组，同一个组里只有一个订阅者会收到该...
Kafka与Redis PUB/SUB之间较大的区别在于Kafka是一个完整的系统，而Redis PUB/SUB只是一个套件(utility)——没有冒犯Redis的意思，毕竟它的主要功能并不是PUB/SUB。redis 消息推送（基于分布式 pub/sub）多用于实时性较高的消息推送，并不保证可靠。（推荐学习：Redis视频教程） 其他的mq和kafka保证可靠但有一些延迟（非实时系统没有保证延迟）。redis-pub/sub断电就清空，而使用redis-list作为消息推送虽然有持久化，但是又太弱...
1，controller层 @Controllerpublic class PlanController { @Autowired private PlanService planService; @Autowired private RedisTemplate redisTemplate; @Autowired private KafkaTemplate<String, String> kafkaTemplate; @Autowired private ElasticsearchTemplate elasticsearchTemplate; @RequestMapping("list") public String list(MultipartFile myFile,Model m,@RequestParam(defaultValue = "1")Integer pageNu...
Kafka作为时下最流行的开源消息系统，被广泛地应用在数据缓冲、异步通信、汇集日志、系统解耦等方面。相比较于RocketMQ等其他常见消息系统，Kafka在保障了大部分功能特性的同时，还提供了超一流的读写性能。 针对Kafka性能方面进行简单分析，相关数据请参考：https://segmentfault.com/a/1190000003985468，下面介绍一下Kafka的架构和涉及到的名词：Topic：用于划分Message的逻辑概念，一个Topic可以分布在多个Broker上。Partition...
kafka版本 1.0.0 spark版本 spark-streaming-kafka-0-10_2.11 class KafkaManagerByRedis(kafkaParams: Map[String, Object]) extends Serializable { private val jedis = JedisUtil.getInstance().getJedis /** * def createDirectStream:InputDStream **/ def createDirectStream[K: ClassTag, V: ClassTag](ssc: StreamingContext, topics: Seq[String]): InputDStream[ConsumerRecord[K, V]] = { //1:readOffse...
前言 一年一度的双11又要到了，阿里的双11销量大屏可以说是一道特殊的风景线。实时大屏（real-time dashboard）正在被越来越多的企业采用，用来及时呈现关键的数据指标。并且在实际操作中，肯定也不会仅仅计算一两个维度。由于Flink的“真流式计算”这一特点，它比Spark Streaming要更适合大屏应用。本文从笔者的实际工作经验抽象出简单的模型，并简要叙述计算流程（当然大部分都是源码）。 数据格式与接入 简化的子订单消息体如下...
CentOS7安装完成后的网络配置 1.输入命令cd /etc/sysconfig/network-scripts/ #进入网络配置文件目录, 2.输入：vi ifcfg-ens33编辑配置文件(此处是ifcfg-ens33,若没有该文件找ifcfg-ens开头的，vi是编辑命令，进入后再按”I”进行编辑，编辑完按“Esc”退出操作并按”:wq”进行保存) 修改内容为： TYPE=Ethernet PROXY_METHOD=none BROWSER_ONLY=no BOOTPROTO=none DEFROUTE=yes IPV4_FAILURE_FATAL=no IPV6INIT=yes I...
Kafka安装 （1）从官方（http://kafka.apache.org/downloads）下载安装包。kafka安装包和一般安装包的命名方式不一样，我们看一个kafka包命名：kafka_2.12-2.1.1.tgz，其中2.12是scala的版本，2.1.1才是kafka的版本。官方强烈建议scala版本和服务器上的scala版本保持一致，避免引发一些不可预知的问题。 （2）执行如下命令解压安装包：bash-4.1$ tar -xzf kafka_2.12-2.1.1.tgz（3）kafka依赖zookeeper，zookeeper集群可以自己搭建...
口述/作者：姜江，新东方信息管理部平台架构负责人陈博暄，新东方信息管理部高级运维工程师编辑：Rancher Labs新东方信息管理部平台架构负责人 姜江新东方信息管理部高级运维工程师 陈博暄2017年，新东方开始了利用容器化手段将中间件业务服务化的探索，基于Rancher 1.6使用ES；2019年，新东方再次开始了扩大了中间件的业务服务化，基于Kubernetes使用Kafka、ES和Redis。利用容器化手段将中间件服务化，有效提升了运维团队的工作效...
从redis中查询月留存率 StatRemainRatioMonth 1 package com.oldboy.umeng.spark.stat;2 3 4 5 /**6 * 统计月留存率7 */8 public class StatRemainRatioMonth {9 public static void main(String[] args) throws Exception { 10 SparkConf conf = new SparkConf(); 11 conf.setAppName("statNew"); 12 conf.setMaster("local[4]"); 13 SparkSession sess = SparkSession.builder().config(c...
实时存储到redis，更新用户使用时间的最大、最小值--》来求留存率 避免全表扫描问题 1.spark 对接kafka 消费者? ?解析json?? 2.concat(appid,'#',appversion,'#',brand,'#',appplatform,'#',devicestyle,'#',ostype,'#',deviceid) ---> 作为key? 各个维度 3.选出用户使用app的最小时间戳min , 最大时间戳max? -->作为value 4.存储到redis ? 依赖 <dependency><groupId>org.apache.spark</groupId><artifactId>spark-stream...