缓存的搜索结果-阿里云开发者社区

首页> 标签> 缓存

"缓存"

共 30433 条结果

全部问答文章公开课课程电子书技术圈体验

Java 并发编程(二)

6、线程的状态流转图线程的生命周期及五种基本状态：7、Java 线程具有五中基本状态1、新建状态（New）：当线程对象对创建后，即进入了新建状态，如：Thread t = new MyThread()； 2、就绪状态（Runnable）：当调用线程对象的 start()方法（t.start();），线程即进入就绪状态。处于就绪状态的线程，只是说明此线程已经做好了准备，随时等待 CPU 调度执行，并不是说执行了 t.start()此线程立即就会执行； 3、运行状态（Running）：当 CPU 开始调度处于就绪状态的线程时，此时线程才得以真正执行，即进入到运行状态。注：就绪状态是进入到运行状态的唯一入口，也就是说，线程要想进入运行状态执行，首先必须处于就绪状态中；4、阻塞状态（Blocked）：处于运行状态中的线程由于某种原因，暂时放弃对 CPU 的使用权，停止执行，此时进入阻塞状态，直到其进入到就绪状态，才有机会再次被 CPU 调用以进入到运行状态。根据阻塞产生的原因不同，阻塞状态又可以分为三种： 1、等待阻塞：运行状态中的线程执行 wait()方法，使本线程进入到等待阻塞状态； 2、同步阻塞：线程在获取 synchronized 同步锁失败(因为锁被其它线程所占用)， 它会进入同步阻塞状态； 3、其他阻塞：通过调用线程的 sleep()或 join()或发出了 I/O 请求时，线程会进入到阻塞状态。当 sleep()状态超时、join()等待线程终止或者超时、或者 I/O 处理完毕时，线程重新转入就绪状态。 5、死亡状态（Dead）：线程执行完了或者因异常退出了 run()方法，该线程结束生命周期。8、什么是线程池？有哪几种创建方式？ 线程池就是提前创建若干个线程，如果有任务需要处理，线程池里的线程就会处理任务，处理完之后线程并不会被销毁，而是等待下一个任务。由于创建和销毁线程都是消耗系统资源的，所以当你想要频繁的创建和销毁线程的时候就可以考虑使用线程池来提升系统的性能。 java 提供了一个 java.util.concurrent.Executor 接口的实现用于创建线程池。9、四种线程池的创建： 1、newCachedThreadPool 创建一个可缓存线程池 2、newFixedThreadPool 创建一个定长线程池，可控制线程最大并发数。 3、newScheduledThreadPool 创建一个定长线程池，支持定时及周期性任务执行。 4、newSingleThreadExecutor 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务。 10、线程池的优点1、重用存在的线程，减少对象创建销毁的开销。 2、可有效的控制最大并发线程数，提高系统资源的使用率，同时避免过多资源竞争，避免堵塞。 3、提供定时执行、定期执行、单线程、并发数控制等功能。

文章

缓存 · Java · 调度

2022-05-21

算法必知 --- LFU缓存淘汰算法

写在前LRU缓存机制（Least Recently Used）（看时间）在缓存满的时候，删除缓存里最久未使用的数据，然后再放入新元素；数据的访问时间很重要，访问时间距离现在越近，就越不容易被删除；就是喜新厌旧，淘汰在缓存里呆的时间最久的元素。在删除元素的时候，只看「时间」这一个维度。LFU缓存机制（Least Frequently Used）（看访问次数）在缓存满的时候，删除缓存里使用次数最少的元素，然后在缓存中放入新元素；数据的访问次数很重要，访问次数越多，就越不容易被删除，即淘汰访问次数最少的；核心思想：先考虑访问次数，在访问次数相同的情况下，再考虑缓存的时间。算法描述请你为最不经常使用（LFU）缓存算法设计并实现数据结构。实现 LFUCache 类：LFUCache(int capacity) - 用数据结构的容量 capacity 初始化对象int get(int key) - 如果键存在于缓存中，则获取键的值，否则返回 -1。void put(int key, int value) - 如果键已存在，则变更其值；如果键不存在，请插入键值对。当缓存达到其容量时，则应该在插入新项之前，使最不经常使用的项无效。在此问题中，当存在平局（即两个或更多个键具有相同使用频率）时，应该去除最近最久未使用的键。注意「项的使用次数」就是自插入该项以来对其调用 get 和 put 函数的次数之和。使用次数会在对应项被移除后置为 0 。为了确定最不常使用的键，可以为缓存中的每个键维护一个使用计数器。使用计数最小的键是最久未使用的键。当一个键首次插入到缓存中时，它的使用计数器被设置为 1 (由于 put 操作)。对缓存中的键执行 get 或 put 操作，使用计数器的值将会递增。注意哦，get 和 put 方法必须都是 O(1) 的时间复杂度！示例：输入： ["LFUCache", "put", "put", "get", "put", "get", "get", "put", "get", "get", "get"] [[2], [1, 1], [2, 2], [1], [3, 3], [2], [3], [4, 4], [1], [3], [4]] 输出： [null, null, null, 1, null, -1, 3, null, -1, 3, 4] 解释： // cnt(x) = 键 x 的使用计数 // cache=[] 将显示最后一次使用的顺序（最左边的元素是最近的） LFUCache lFUCache = new LFUCache(2); lFUCache.put(1, 1); // cache=[1,_], cnt(1)=1 lFUCache.put(2, 2); // cache=[2,1], cnt(2)=1, cnt(1)=1 lFUCache.get(1); // 返回 1 // cache=[1,2], cnt(2)=1, cnt(1)=2 lFUCache.put(3, 3); // 去除键 2 ，因为 cnt(2)=1 ，使用计数最小 // cache=[3,1], cnt(3)=1, cnt(1)=2 lFUCache.get(2); // 返回 -1（未找到） lFUCache.get(3); // 返回 3 // cache=[3,1], cnt(3)=2, cnt(1)=2 lFUCache.put(4, 4); // 去除键 1 ，1 和 3 的 cnt 相同，但 1 最久未使用 // cache=[4,3], cnt(4)=1, cnt(3)=2 lFUCache.get(1); // 返回 -1（未找到） lFUCache.get(3); // 返回 3 // cache=[3,4], cnt(4)=1, cnt(3)=3 lFUCache.get(4); // 返回 4 // cache=[3,4], cnt(4)=2, cnt(3)=3算法设计首先需要维护一个链表，链表的结构如下。ps：注意是双端链表，图示有误，但意思明确，具体见代码。注意：首先我们在LRU中定义两个永久节点，head作为头节点、tail作为尾节点，就作为了我们链表的头节点和尾节点。我们插入的每一个缓存都是链表中的一个Node节点。因为链表频数大的靠近head，频数小的靠近tail。这种情况实际上是将我们的链表按照频数划分成了不同的区域如下图算法需要实现的三个操作：新增节点：如果新增节点的频数为1，所以我们需要找到当前链表频数为1的部分的第一个节点（头结点），在他前面插入新元素(如果不存在频数为1那么就是在tail节点前插入)修改节点（删除+移动）：首先需要将节点的值进行修改这个很简单，然后就是移动节点在链表中的位置了，假设节点的频数为 a ，那么节点首先需要从频数为a的区域中删除，这就分为了以下几种情况：频数为a的区域只有一个节点，那么a节点频数修改后，频数为a的区域将会消失(注意这个说法，后面会讲实现)，然后将当前暂时从链表中移除频数为a的区域的头节点是当前节点，那么将频数为a的节点的头节点改为当前节点的后一个元素即可，然后将当前节点暂时从链表中移除频数为a的区域的头节点不是当前节点，那么直接将当前节点暂时从频数为a的区域删除即可。从频数为a的区域删除后，下面一步就是插入到频数为a+1的区域的头部：频数为a+1的区域不存在，那么将当前节点插入到上一步更新后的频数为a的头节点的前面即可频数为a+1的区域存在，直接插入到频数为a+1的区域的头部即可删除节点：因为tail节点的前一个就是我们使用次数最少且最不常使用的缓存，我们直接删除tail节点的前一个节点即可，单数删除的时候需要注意。假设需要删除的节点的频数为a，这个操作相当于将指定节点从频数为a的区域删除，这和我们修改阶段的第一步是类似的。总结：上述算法实现的重点，如何获得频数为a的区域的头节点（使用一个Map来维护链表中频数为a的区域的头节点）。代码实现首先定义双端链表类（包括数据和记录前驱/后继节点的指针）class DLinkedNode { int key; int value; // 记录当前key被调用的次数 int count; DLinkedNode pre; DLinkedNode next; public DLinkedNode() {}; public DLinkedNode(int key, int value) { this.key = key; this.value = value; this.count = count; } }双向链表需要提供一些接口api，便于我们操作，主要就是链表的一些操作，画图理解！private void renewNode(DLinkedNode node) { int oldCnt = node.count; int newCnt = oldCnt + 1; DLinkedNode next = null; if (cntMap.get(oldCnt) == node) { //当前节点是oldCnt频数的头结点（两种情况：还有其他节点/只有一个节点） // 更新oldCnt频数头结点的映射 if (node.next.count == node.count) { cntMap.put(oldCnt, node.next); } else { cntMap.remove(oldCnt); } // 更新newCnt频数头结点的映射（不存在直接加入，存在找到对应频数的头结点） if (cntMap.get(newCnt) == null) { cntMap.put(newCnt, node); node.count++; return; } else { removeFromList(node); next = cntMap.get(newCnt); } } else { // 当前节点不是某个频数的头结点（我们不需要维护频数头结点的映射，直接找到对应频数的头结点即可） removeFromList(node); if (cntMap.get(newCnt) == null) { next = cntMap.get(oldCnt); } else { next = cntMap.get(newCnt); } } node.count++; cntMap.put(newCnt, node); // 插入节点（连接节点），其中next是频数的头结点 insertToList(node, next); } private void removeFromList(DLinkedNode node) { node.pre.next = node.next; node.next.pre = node.pre; } private void insertToList(DLinkedNode node, DLinkedNode next) { next.pre.next = node; node.pre = next.pre; node.next = next; next.pre = node; } // 缓存容量满了，删除一个最少且最久没使用的节点 private void deleteCache() { DLinkedNode delNode = tail.pre; DLinkedNode pre = delNode.pre; if (cntMap.get(delNode.count) == delNode) { // 删除节点是某个频数的头结点 cntMap.remove(delNode.count); } // 实际删除的节点 pre.next = tail; tail.pre = pre; cache.remove(delNode.key); --size; }确定LRU缓存类的成员变量（链表长度、缓存容量和map映射等）和构造函数。注意：定义虚拟头尾结点便于在头部插入元素或者寻找尾部元素！并在构造函数初始化。// cnt - node : 增加了频数与头节点的映射 private Map<Integer, DLinkedNode> cntMap = new HashMap<>(); // key - node private Map<Integer, DLinkedNode> cache = new HashMap<>(); // 缓存中目前存储的数据量 private int size; private int capacity; private DLinkedNode head, tail; public LFUCache(int capacity) { this.size = 0; this.capacity = capacity; head = new DLinkedNode(); tail = new DLinkedNode(); head.next = tail; tail.pre = head; }核心代码：get和put方法，都是先根据key获取这个映射，根据映射节点的情况（有无）进行操作。注意：get和put都在使用，所以数据要提前！put操作如果改变了双端链表长度（不是仅改变值），需要先判断是否达到最大容量！public int get(int key) { DLinkedNode node = cache.get(key); if (capacity == 0 || node == null) { return -1; } // node节点的调用次数+1,应该更新他的位置 renewNode(node); return node.value; } public void put(int key, int value) { if (capacity == 0) { return; } DLinkedNode node = cache.get(key); if (node != null) { node.value = value; // 将这个节点的频数cnt+1，更新位置 renewNode(node); } else { if (cache.size() == capacity) { deleteCache(); --size; } DLinkedNode newNode = new DLinkedNode(key, value, 1); DLinkedNode next = cntMap.get(1); if (next == null) { next = tail; } // 将新建的节点插入到链表，并更新映射 insertToList(newNode, next); cntMap.put(1, newNode); cache.put(key, newNode); ++size; } }完整代码如下：class LFUCache { class DLinkedNode { int key; int value; // 记录当前key被调用的次数(即node节点的频数) int count; DLinkedNode pre; DLinkedNode next; public DLinkedNode() {}; public DLinkedNode(int key, int value, int count) { this.key = key; this.value = value; this.count = count; } } // cnt - node : 增加了频数与头节点的映射 private Map<Integer, DLinkedNode> cntMap = new HashMap<>(); // key - node private Map<Integer, DLinkedNode> cache = new HashMap<>(); // 缓存中目前存储的数据量 private int size; private int capacity; private DLinkedNode head, tail; public LFUCache(int capacity) { this.size = 0; this.capacity = capacity; head = new DLinkedNode(); tail = new DLinkedNode(); head.next = tail; tail.pre = head; } public int get(int key) { DLinkedNode node = cache.get(key); if (capacity == 0 || node == null) { return -1; } // node节点的调用次数+1,应该更新他的位置 renewNode(node); return node.value; } public void put(int key, int value) { if (capacity == 0) { return; } DLinkedNode node = cache.get(key); if (node != null) { node.value = value; // 将这个节点的频数cnt+1，更新位置 renewNode(node); } else { if (cache.size() == capacity) { deleteCache(); --size; } DLinkedNode newNode = new DLinkedNode(key, value, 1); DLinkedNode next = cntMap.get(1); if (next == null) { next = tail; } // 将新建的节点插入到链表，并更新映射 insertToList(newNode, next); cntMap.put(1, newNode); cache.put(key, newNode); ++size; } } private void renewNode(DLinkedNode node) { int oldCnt = node.count; int newCnt = oldCnt + 1; DLinkedNode next = null; if (cntMap.get(oldCnt) == node) { //当前节点是oldCnt频数的头结点（两种情况：还有其他节点/只有一个节点） // 更新oldCnt频数头结点的映射 if (node.next.count == node.count) { cntMap.put(oldCnt, node.next); } else { cntMap.remove(oldCnt); } // 更新newCnt频数头结点的映射（不存在直接加入，存在找到对应频数的头结点） if (cntMap.get(newCnt) == null) { cntMap.put(newCnt, node); node.count++; return; } else { removeFromList(node); next = cntMap.get(newCnt); } } else { // 当前节点不是某个频数的头结点（我们不需要维护频数头结点的映射，直接找到对应频数的头结点即可） removeFromList(node); if (cntMap.get(newCnt) == null) { next = cntMap.get(oldCnt); } else { next = cntMap.get(newCnt); } } node.count++; cntMap.put(newCnt, node); // 插入节点（连接节点），其中next是频数的头结点 insertToList(node, next); } private void removeFromList(DLinkedNode node) { node.pre.next = node.next; node.next.pre = node.pre; } private void insertToList(DLinkedNode node, DLinkedNode next) { next.pre.next = node; node.pre = next.pre; node.next = next; next.pre = node; } // 缓存容量满了，删除一个最少且最久没使用的节点 private void deleteCache() { DLinkedNode delNode = tail.pre; DLinkedNode pre = delNode.pre; if (cntMap.get(delNode.count) == delNode) { // 删除节点是某个频数的头结点 cntMap.remove(delNode.count); } // 实际删除的节点 pre.next = tail; tail.pre = pre; cache.remove(delNode.key); --size; } }总结与补充上述代码在LRU基础上进行的。主要区别是：节点类中引入了count变量，记录key出现的频数LFU成员变量中增加了cntMap：key的频数与这个频数区间头结点的映射注意：同样的，我们要注意维护cntMap映射和节点的频数

文章

缓存 · 算法 · API

2022-05-21

数据结构与算法必知--- Bitmap位图与布隆过滤器

写在前bitmap和布隆过滤器主要解决大数据去重的问题。用于对大量整型数据做去重和查询。其实如果并非如此大量的数据，有很多排重方案可以使用，典型的就是哈希表。实际上，哈希表为每一个可能出现的数字提供了一个一一映射的关系，每个元素都相当于有了自己的独享的一份空间，这个映射由散列函数来提供（这里我们先不考虑碰撞）。实际上哈希表甚至还能记录每个元素出现的次数，这样的数据结构完成这个任务有点“大材小用”了。如果用HashSet或HashMap存储，每一个用户ID都要存成int，占4字节即32bit。而一个用户在Bitmap中只占一个bit，内存节省了32倍！所以散列表最大的问题是所占用的空间非常大！！！Bitmap位图算法为什么用位图？我们先来看一个问题，假设我们有1千万个（不重复、未排序）的整数需要存储，每个整数的大小范围是1到1亿。然后，给定任意一个整数X，我们需要快速判断X是否在刚才的1千万个整数内。这个问题该如何处理呢？常规的做法肯定就是先考虑如何存储这1千万个整数，在Java中，int类型是4个字节，可以表示的范围区间是-2147483648~2147483647，所以每个整数都用int来表示是可行的。那么1千万个整数需要占用多少内存空间呢？两者相乘就是了，应该是4000万字节，也就是40MB。而且如果我们存放的整数是是一亿个（甚至更大），每个整数大小范围是1到100亿怎么办？占用的内存空间将会是非常巨大的，是普通的计算机或者手机不能接受的（放不下）。ps：多台机器存储的话，资源占用量太大；另外可以不用内存，但是I/O效率低；所以我们需要使用位图这种数据结构来大大节省内存的使用量。什么是位图？我们知道Byte表示字节，一个字节等于8bit，这里的bit就和位图有关系。在上面的例子中，我们不是要存放1千万个整数嘛，那就申请一个具有1千万个bit的数组，用每个bit（二进制位）来表示一个整数，当前bit为0表示不存在这个整数，为1表示该整数就存在。虽然数量是1千万个，但是每个数的范围是1到1亿，所以我们需要1亿个bit，换算下来，就是12.5MB，和刚才的40MB相比，节省的不是一点点。位图是指内存中连续的二进制位，用于对大量的整型数据做去重和查询。Bit-map就是用一个bit位来标记某个元素对应的Value，而Key即是该元素。对于位图的底层存储结构实际为数组结构,目的是为了存储大量的数据,为了表示存储的数据是否存在,可以使用0/1代表true/false。存储的数据流程为根据自定义的hash计算公式得到对应的数组下标并将数据进行相应的位运算得到的结果存储到数组中;当需要查询数据时将当前要查询的数据转换为对应的数组索引根据索引定位数组中的数据并执行存储时位运算的相反操作判断数据是否存在可以看出位图基于数组下标查询效率非常高效,且相对而言,位运算使用cpu计算也比较高效；由于采用了Bit为单位来存储数据，因此在存储空间方面，可以大大节省。位图也存在一些缺点,对于随着数据量的增加,要申请数组的空间也越来越大。bitmap应用1）可进行数据的快速查找，判重，删除，一般来说数据范围是int的10倍以下。2）去重数据而达到压缩数据比如 Java 中的 BitSet 类就是一个位图，Redis 也提供了 BitMap 位图类布隆过滤器为什么用布隆过滤器？位图存在两个问题处理数字很方便，但是处理字符就稍微有点困难了（关键）位图在数据量过大情况下导致了内存浪费布隆过滤器为了解决内存占用的问题,由于位图实际是利用一个index来定位一个数据,如果在数据密集度相对稀疏的情况下,会导致数组空间浪费(存在大量空洞插槽); 布隆过滤器利用多个插槽位置来定义一个元素,尽最大的可能性来提高空间利用率; 对于一个数据对应多个插槽,实际就是利用多个hash函数生成多个索引位置。什么是布隆过滤器？布隆过滤器实际是基于位图来实现的一种数据存储结构，用来判断某个元素（key）是否在某个集合中。和一般的位图不同的是，这个算法无需存储key的值，对于每个key，只需要k个比特位，每个存储一个标志，用来判断key是否在集合中。布隆过滤器最大的特点，就是对一个对象使用多个哈希函数。如果我们使用了 k 个哈希函数，就会得到 k 个哈希值，也就是 k 个下标，我们会把数组中对应下标位置的值都置为 1。布隆过滤器和位图最大的区别就在于，我们不再使用一位来表示一个对象，而是使用 k 位来表示一个对象。这样两个对象的 k 位都相同的概率就会大大降低，不仅能够解决单哈希容易造成冲突的问题，还能有效减少内存空间的使用。但要注意的是,布隆过滤器主要是为了解决空间浪费问题,并不能完全解决hash冲突;由于hash冲突的存在会导致一些误判的发生（无中生有）,虽然使用多个hash函数可以降低冲突的概率,但还是会存在一些极端的情况出现hash冲突,导致不存在的数据被布隆过滤器判定为存在;但是对于布隆过滤器判定为不存在的数据,是肯定不存在的。虽然布隆过滤器对于空间利用率较高,但随着数组剩余空间越来越小,会导致hash冲突的概率越来越大,因此需要设置一个阈值,当剩余空间达到一定阈值后,就需要执行扩容,降低hash冲突概率。优缺点：优点：不需要存储key，节省空间缺点：算法判断key在集合中时，有一定的概率key其实不在集合中，已经映射的数据无法删除应用场景redis 缓存穿透就可以利用布隆过滤器来降低缓存穿透发生概率;redis 缓存穿透的理解 , 对于 redis中和数据库中都不存在的数据,当从redis中获取不到数据时,请求就会请求到服务端执行db查询操作（绕过缓存）,当存在大量缓存穿透的情况就会导致redis和服务以及数据库的压力剧增;当redis 使用布隆过滤器后,首先能通过布隆过滤器的判定,对于不存在的数据,redis和db中肯定都不存在;但对于布隆过滤器判断为存在的数据,即使经过redis查询发生数据不存在,从而查找db也发现数据不存在的情况, 但对于这种情况毕竟几率较低,即使出现了对整个redis以及服务的压力也不会非常大其他：比如网络爬虫抓取时url去重，邮件提供商反垃圾黑名单Email地址去重，之所以需要k个比特位是因为我们大多数情况下处理的是字符串，那么不同的字符串就有可能映射到同一个位置，产生冲突（尽可能的降低hash冲突）。总结BitMap:把数值转化为数组的下标，下标对应的值只存放一个bit 0或1 ，0不存在 1存在。判断时根据数值当做下标找到对应的 bit 值，1存在，0不存在BloomFilter:把内容通过多个hash算法，转换成多个数组下标，下标对应的值只存放一个bit 0或1 ，0不存在 1存在。判断过滤时多个下标对应的值都为1，则可能存在，只要有一个值为0，则一定不存在

文章

存储 · 数据采集 · 缓存 · NoSQL · 算法 · Java · 大数据 · Redis · 数据库 · 索引

2022-05-21

Java 并发编程

6、什么是线程组，为什么在 Java 中不推荐使用？ ThreadGroup 类，可以把线程归属到某一个线程组中，线程组中可以有线程对象， 也可以有线程组，组中还可以有线程，这样的组织结构有点类似于树的形式。为什么不推荐使用？因为使用有很多的安全隐患吧，没有具体追究，如果需要使用，推荐使用线程池。7、为什么使用 Executor 框架？ 每次执行任务创建线程 new Thread()比较消耗性能，创建一个线程是比较耗时、耗资源的。调用 new Thread()创建的线程缺乏管理，被称为野线程，而且可以无限制的创建， 线程之间的相互竞争会导致过多占用系统资源而导致系统瘫痪，还有线程之间的频繁交替也会消耗很多系统资源。接使用 new Thread() 启动的线程不利于扩展，比如定时执行、定期执行、定时定期执行、线程中断等都不便实现。 8、在 Java 中 Executor 和 Executors 的区别？ Executors 工具类的不同方法按照我们的需求创建了不同的线程池，来满足业务的需求。 Executor 接口对象能执行我们的线程任务。 ExecutorService 接口继承了 Executor 接口并进行了扩展，提供了更多的方法我们能获得任务执行的状态并且可以获取任务的返回值。使用 ThreadPoolExecutor 可以创建自定义线程池。Future 表示异步计算的结果，他提供了检查计算是否完成的方法，以等待计算的完成，并可以使用 get()方法获取计算的结果。9、如何在 Windows 和 Linux 上查找哪个线程使用的 CPU 时间最长？ 参考： http://daiguahub.com/2016/07/31/ 使用 jstack 找出消耗 CPU 最多的线程代码 / 10、什么是原子操作？在 Java Concurrency API 中有哪些原子类(atomic classes)？ 原子操作（atomic operation）意为”不可被中断的一个或一系列操作” 。处理器使用基于对缓存加锁或总线加锁的方式来实现多处理器之间的原子操作。在 Java 中可以通过锁和循环 CAS 的方式来实现原子操作。 CAS 操作—— Compare & Set，或是 Compare & Swap，现在几乎所有的 CPU 指令都支持 CAS 的原子操作。原子操作是指一个不受其他操作影响的操作任务单元。原子操作是在多线程环境下避免数据不一致必须的手段。 int++并不是一个原子操作，所以当一个线程读取它的值并加 1 时，另外一个线程有可能会读到之前的值，这就会引发错误。为了解决这个问题，必须保证增加操作是原子的，在 JDK1.5 之前我们可以使用同步技术来做到这一点。到 JDK1.5，java.util.concurrent.atomic 包提供了 int 和 long 类型的原子包装类，它们可以自动的保证对于他们的操作是原子的并且不需要使用同步。java.util.concurrent 这个包里面提供了一组原子类。其基本的特性就是在多线程环境下，当有多个线程同时执行这些类的实例包含的方法时，具有排他性，即当某个线程进入方法，执行其中的指令时，不会被其他线程打断，而别的线程就像自旋锁一样，一直等到该方法执行完成，才由 JVM 从等待队列中选择一个另一个线程进入，这只是一种逻辑上的理解。原子类：AtomicBoolean，AtomicInteger，AtomicLong，AtomicReference 原子数组：AtomicIntegerArray，AtomicLongArray，AtomicReferenceArray 原子属性更新器：AtomicLongFieldUpdater，AtomicIntegerFieldUpdater， AtomicReferenceFieldUpdater 解决 ABA 问题的原子类：AtomicMarkableReference（通过引入一个 boolean 来反映中间有没有变过），AtomicStampedReference（通过引入一个 int 来累加来反映中间有没有变过）

文章

缓存 · Java · Linux · API · Windows

2022-05-21

NVIDIA Tesla GPU系列P4、T4、P40以及V100参数性能对比

? 一、关于NVIDIA TESLA系列GPU详细介绍如下： NVIDIA TESLA V100 NVIDIA Tesla V100采用NVIDIA Volta架构，非常适合为要求极为苛刻的双精度计算工作流程提供加速，并且还是从P100升级的理想路径。该GPU的渲染性能比Tesla P100提升了高达80%，借此可缩短设计周期和上市时间。 Tesla V100的每个GPU均可提供125 teraflops的推理性能，配有8块Tesla V100的单个服务器可实现1 petaflop的计算性能。 NVIDIA TESLA P40 The Tesla P40能够提供高达2倍的专业图形性能。Tesla P40能够对组织中每个vGPU虚拟化加速图形和计算（NVIDIA CUDA? 和 OpenCL）工作负载。支持多种行业标准的2U服务器。 Tesla P40可提供出色的推理性能、INT8精度和24GB板载内存。 NVIDIA TESLA T4 NVIDIA Tesla T4的帧缓存高达P4的2倍，性能高达M60的2倍，对于利用NVIDIA Quadro vDWS软件开启高端3D设计和工程工作流程的用户而言，不失为一种理想的解决方案。凭借单插槽、半高外形特性以及低至70瓦的功耗，Tesla T4堪称为每个服务器节点实现最大GPU密度的绝佳之选。 NVIDIA TESLA P4 Tesla P4可加快任何外扩型服务器的运行速度，能效高达CPU的60倍。 ? 二、NVIDIA Tesla GPU系列P4、T4、P40以及V100 NVIDIA Tesla系列GPUP4、T4、P40以及V100性能规格参数对比表，阿里云GPU云服务器提供的实例GN4（Nvidia M40）、GN5（Nvidia P100）、GN5i（Nvidia P4）及GN6（Nvidia V100），也会基于NVIDIA Tesla GPU系列。云服务器 Tesla T4：世界领先的推理加速器 Tesla V100：通用数据中心 GPU 适用于超高效、外扩型服务器的 Tesla P4 适用于推理吞吐量服务器的 Tesla P40 单精度性能 (FP32) 8.1 TFLOPS 14 TFLOPS (PCIe) 15.7 teraflops (SXM2) 5.5 TFLOPS 12 TFLOPS 半精度性能 (FP16) 65 TFLOPS 112 TFLOPS (PCIe)125 TFLOPS (SXM2) — — 整数运算能力 (INT8) 130 TOPS — 22 TOPS* 47 TOPS* 整数运算能力 (INT4) 260 TOPS — — — GPU 显存 16GB 32/16GB HBM2 8GB 24GB 显存带宽 320GB/秒 900GB/秒 192GB/秒 346GB/秒系统接口/外形规格 PCI Express 半高外形 PCI Express 双插槽全高外形 SXM2/NVLink PCI Express 半高外形 PCI Express 双插槽全高外形功率 70 W 250 W (PCIe) 300 W (SXM2) 50 W/75 W 250 W 硬件加速视频引擎 1 个解码引擎，2 个编码引擎 — 1 个解码引擎，2 个编码引擎 1 个解码引擎，2 个编码引擎 ? ?

文章

机器学习/深度学习 · 缓存 · 并行计算 · openCL · 虚拟化 · 数据中心 · 异构计算

2022-05-21

Laravel 7发行说明（二）

路由模型绑定优化路由模型绑定优化由 Taylor Otwell 开发贡献。自定义键名有时你可能希望使用 id 以外的字段来解析 Eloquent 模型。为此， Laravel 7 允许你在路由参数中指定某个字段：Route::get('api/posts/{post:slug}', function (App\Post $post) { return $post; });隐式绑定约束有时，当在路由中隐式绑定多个 Eloquent 模型时，可能希望对第二个 Eloquent 模型进行约束，使其必须是第一个 Eloquent 模型的子类。例如，考虑这种情况，该情况是通过 Slug 为特定用户查找博客文章的：use App\Post; use App\User; Route::get('api/users/{user}/posts/{post:slug}', function (User $user, Post $post) { return $post; });当使用自定义键隐式绑定作为嵌套的路由参数时，Laravel 7 将自动确定查询范围，以使用约定猜测其父级上的关系名称，以其父级检索嵌套模型。在这种情况下，将假定 User 模型关联了名为 posts(路由参数名称的复数) 的关系，该关系可用于检索 Post 模型。有关路由模型绑定的更多信息，请查阅路由文档。多邮件驱动程序多邮件驱动程序支持由 Taylor Otwell 贡献。Laravel 7 允许为单个应用配置多个邮件驱动。在 mail 配置文件中的每个邮件驱动都拥有它们自己的配置以及自己独特的「transport」，这允许你的应用使用不同的邮件服务来发送某些邮件。例如，你的应用可以使用 Postmark 发送批量邮件，使用 Amazon SES 发送公务邮件。默认情况下，Laravel 将使用 mail 配置文件中的 default 选项指定的邮件驱动作为邮件驱动。然而，你可以通过 mailer 方法来使用特定的邮件驱动来发送邮件。Mail::mailer('postmark') ->to($request->user()) ->send(new OrderShipped($order));路由缓存速度改进路由缓存速度改进由上游的 Symfony 的贡献者和 Dries Vints 贡献Laravel 7 提供了一种新的方法，用于匹配使用 Artisan 命令 route：cache 缓存的已编译缓存路由。在大型应用程序（例如，具有800条或更多路由的应用程序）上，这些改进可以使简单的「Hello World」基准测试每秒的请求速度提高2倍 ，而无需更改应用程序。CORS 支持CORS 支持由 Barry vd. Heuvel 贡献Laravel 7 通过集成由 Barry vd. Heuvel 编写的受欢迎的 Laravel CORS 软件包，为配置跨域资源共享（CORS） OPTIONS 请求响应提供了官方支持， 默认的 Laravel 应用程序框架中包含一个新的 cors 配置。有关 Laravel 7.x 中的 CORS 支持的更多信息，请查阅CORS文档。查询时类型转换查询时类型转换由 Matt Barlow 开发贡献.有时候需要在查询执行过程中对特定属性进行类型转换，例如需要从数据库表中获取数据的时候。举个例子，请参考以下查询：use App\Post; use App\User; $users = User::select([ 'users.*', 'last_posted_at' => Post::selectRaw('MAX(created_at)') ->whereColumn('user_id', 'users.id') ])->get();在该查询获取到的结果集中，last_posted_at 属性将会是一个字符串。假如我们在执行查询时进行 date 类型转换将更方便。你可以通过使用 withCasts 方法来完成上述操作:$users = User::select([ 'users.*', 'last_posted_at' => Post::selectRaw('MAX(created_at)') ->whereColumn('user_id', 'users.id') ])->withCasts([ 'last_posted_at' => 'date' ])->get();MySQL 8+ 数据库队列改进MySQL 数据库队列改进由 Mohamed Said 开发贡献.在先前版本的 Laravel 中， database 队列的健壮性被认为无法满足生产环境的需求。但是，Laravel 7 针对使用基于 MySQL 8+ 数据库队列的应用进行了改进。通过使用 FOR UPDATE SKIP LOCKED 语句进行 SQL 的优化，database 队列驱动可以安全地用于生产环境。Artisan test 命令test 命令由 Nuno Maduro 贡献除了 phpunit命令之外，现在可以使用 test Artisan 命令来运行测试。 Artisan 测试运行器提供了漂亮的控制台，以及有关当前正在运行的测试的更多信息。此外，运行器将在第一次测试失败时自动停止：php artisan test可以传递给 phpunit 命令的任何参数也可以传递给 Artisan test 命令：php artisan test --group=featureMarkdown 邮件模板改进Markdown 邮件模板改进由 Taylor Otwell 贡献默认的Markdown邮件模板已基于Tailwind CSS调色板做出全新、更现代的设计。当然，可以根据您的应用程序的需求来发布和定制此模板：有关 Markdown 邮件的更多信息，请查看邮件发送.自定义桩代码自定义桩代码由 Taylor Otwell贡献Artisan 控制台的 make 命令用于创建各种类，例如控制器，任务，迁移和测试。这些类是根据输入填充值使用「桩代码」生成文件的。但是，有时可能希望对 Artisan 生成的文件进行小的更改。为此，Laravel 7提供了 stub：publish 命令来发布最常见的自定义桩代码：php artisan stub:publish发布的桩代码将位于应用程序根目录中的 stubs 目录中。当使用 Artisan 的 make 命令生成它们的相应类时，对这些桩代码所做的任何更改都会反映出来。队列maxExceptions配置maxExceptions属性由Mohamed Said提交贡献.有时可能希望指定可以尝试多次的任务，但是如果重试是由给定数量的异常触发的，则该任务将失败。在Laravel7中，可以在任务类上定义 maxExceptions 属性：<?php namespace App\Jobs; class ProcessPodcast implements ShouldQueue { /** * 任务可以被重试的次数。 * * @var int */ public $tries = 25; /** * 失败之前允许抛出异常的最大次数。 * * @var int */ public $maxExceptions = 3; /** * 执行任务。 * * @return void */ public function handle() { Redis::throttle('key')->allow(10)->every(60)->then(function () { // 获取锁，处理博客进程... }, function () { // 无法获取锁... return $this->release(10); }); } }在此示例中，如果应用程序无法获得 Redis 锁，则该任务将释放十秒钟，并将继续重试 25 次。但是，如果任务抛出三个未处理的异常，则该任务将失败。

文章

SQL · 缓存 · 关系型数据库 · MySQL · 测试技术 · PHP · 数据库 · 前端开发 · NoSQL · Redis

2022-05-18

客快物流大数据项目(六十七)：客户主题（一）

客户主题一、背景介绍客户主题主要是通过分析用户的下单情况构建用户画像二、指标明细三、表关联关系1、事实表2、维度表3、关联关系用户表与维度表的关联关系如下：四、客户数据拉宽开发1、拉宽后的字段2、SQL语句SELECT TC."id" , TC."name" , TC."tel", TC."mobile", TC."email", TC."type", TC."is_own_reg", TC."reg_dt", TC."reg_channel_id", TC."state", TC."cdt", TC."udt", TC."last_login_dt", TC."remark", customercodes."code_desc", sender_info.first_cdt AS first_sender_cdt , sender_info.last_cdt AS last_sender_cdt, sender_info.billCount AS billCount, sender_info.totalAmount AS totalAmount FROM "tbl_customer" tc LEFT JOIN ( SELECT "ciid", min(sender_info."id") first_id, max(sender_info."id") last_id, min(sender_info."cdt") first_cdt, max(sender_info."cdt") last_cdt,COUNT(sender_info."id" ) billCount,sum(express_package."actual_amount") totalAmount FROM "tbl_consumer_sender_info" sender_info LEFT JOIN "tbl_express_package" express_package ON SENDER_INFO."pkg_id" =express_package."id" GROUP BY sender_info."ciid" ) sender_info ON tc."id" = sender_info."ciid" LEFT JOIN "tbl_codes" customercodes ON customercodes."type" =16 AND tc."type" =customercodes."code"3、Spark实现实现步骤：在dwd目录下创建 CustomerDWD 单例对象，继承自OfflineApp特质初始化环境的参数，创建SparkSession对象获取客户表（tbl_customer）数据，并缓存数据判断是否是首次运行，如果是首次运行的话，则全量装载数据（含历史数据）获取客户寄件信息表（tbl_consumer_sender_info）数据，并缓存数据获取客户包裹表（tbl_express_package）数据，并缓存数据获取物流字典码表（tbl_codes）数据，并缓存数据根据以下方式拉宽仓库车辆明细数据根据客户id，在客户表中获取客户数据根据包裹id，在包裹表中获取包裹数据根据客户类型id，在物流字典码表中获取客户类型名称数据创建客户明细宽表（若存在则不创建）将客户明细宽表数据写入到kudu数据表中删除缓存数据3.1、初始化环境变量初始化客户明细拉宽作业的环境变量package cn.it.logistics.offline.dwd import cn.it.logistics.common.{CodeTypeMapping, Configuration, OfflineTableDefine, SparkUtils} import cn.it.logistics.offline.OfflineApp import org.apache.spark.SparkConf import org.apache.spark.sql.{DataFrame, SparkSession} import org.apache.spark.storage.StorageLevel import org.apache.spark.sql.functions._ import org.apache.spark.sql.types.IntegerType /** * 客户主题数据的拉宽操作 */ object CustomerDWD extends OfflineApp { //定义应用的名称 val appName = this.getClass.getSimpleName def main(args: Array[String]): Unit = { /** * 实现步骤： * 1）初始化sparkConf对象 * 2）创建sparkSession对象 * 3）加载kudu中的事实表和维度表的数据（将加载后的数据进行缓存） * 4）定义维度表与事实表的关联 * 5）将拉宽后的数据再次写回到kudu数据库中（DWD明细层） * 5.1：创建车辆明细宽表的schema表结构 * 5.2：创建车辆宽表（判断宽表是否存在，如果不存在则创建） * 5.3：将数据写入到kudu中 * 6）将缓存的数据删除掉 * 7）停止任务 */ //1）初始化sparkConf对象 val sparkConf: SparkConf = SparkUtils.autoSettingEnv( SparkUtils.sparkConf(appName) ) //2）创建sparkSession对象 val sparkSession: SparkSession = SparkUtils.getSparkSession(sparkConf) sparkSession.sparkContext.setLogLevel(Configuration.LOG_OFF) //数据处理 execute(sparkSession) } /** * 数据处理 * * @param sparkSession */ override def execute(sparkSession: SparkSession): Unit = { sparkSession.stop() } }3.2、加载客户相关的表并缓存加载客户表的时候，需要指定日期条件，因为客户主题最终需要Azkaban定时调度执行，每天执行一次增量数据，因此需要指定日期。判断是否是首次运行，如果是首次运行的话，则全量装载数据（含历史数据）//导入隐士转换 import sparkSession.implicits._ val customerSenderInfoDF: DataFrame = getKuduSource(sparkSession, TableMapping.consumerSenderInfo, Configuration.isFirstRunnable).persist(StorageLevel.DISK_ONLY_2) val customerDF = getKuduSource(sparkSession, TableMapping.customer, true).persist(StorageLevel.DISK_ONLY_2) val expressPageageDF = getKuduSource(sparkSession, TableMapping.expressPackage, true).persist(StorageLevel.DISK_ONLY_2) val codesDF: DataFrame = getKuduSource(sparkSession, TableMapping.codes, true).persist(StorageLevel.DISK_ONLY_2) val customerTypeDF = codesDF.where($"type" === CodeTypeMapping.CustomType)3.3、定义表的关联关系为了在DWS层任务中方便的获取每日增量客户表数据（根据日期），因此在DataFrame基础上动态增加列（day），指定日期格式为yyyyMMdd代码如下：//TODO 4）定义维度表与事实表的关联关系 val left_outer = "left_outer" /** * 获取每个用户的首尾单发货信息及发货件数和总金额 */ val customerSenderDetailInfoDF: DataFrame = customerSenderInfoDF.join(expressPageageDF, expressPageageDF("id") === customerSenderInfoDF("pkgId"), left_outer) .groupBy(customerSenderInfoDF("ciid")) .agg(min(customerSenderInfoDF("id")).alias("first_id"), max(customerSenderInfoDF("id")).alias("last_id"), min(expressPageageDF("cdt")).alias("first_cdt"), max(expressPageageDF("cdt")).alias("last_cdt"), count(customerSenderInfoDF("id")).alias("totalCount"), sum(expressPageageDF("actualAmount")).alias("totalAmount") ) val customerDetailDF: DataFrame = customerDF .join(customerSenderDetailInfoDF, customerDF("id") === customerSenderInfoDF("ciid"), left_outer) .join(customerTypeDF, customerDF("type") === customerTypeDF("code").cast(IntegerType), left_outer) .sort(customerDF("cdt").asc) .select( customerDF("id"), customerDF("name"), customerDF("tel"), customerDF("mobile"), customerDF("type").cast(IntegerType), customerTypeDF("codeDesc").as("type_name"), customerDF("isownreg").as("is_own_reg"), customerDF("regdt").as("regdt"), customerDF("regchannelid").as("reg_channel_id"), customerDF("state"), customerDF("cdt"), customerDF("udt"), customerDF("lastlogindt").as("last_login_dt"), customerDF("remark"), customerSenderDetailInfoDF("first_id").as("first_sender_id"), //首次寄件id customerSenderDetailInfoDF("last_id").as("last_sender_id"), //尾次寄件id customerSenderDetailInfoDF("first_cdt").as("first_sender_cdt"), //首次寄件时间 customerSenderDetailInfoDF("last_cdt").as("last_sender_cdt"), //尾次寄件时间 customerSenderDetailInfoDF("totalCount"), //寄件总次数 customerSenderDetailInfoDF("totalAmount") //总金额 )3.4、创建客户明细宽表并将客户明细数据写入到kudu数据表中客户明细宽表数据需要保存到kudu中，因此在第一次执行客户明细拉宽操作时，客户明细宽表是不存在的，因此需要实现自动判断宽表是否存在，如果不存在则创建实现步骤：在CustomerDWD 单例对象中调用save方法实实现过程：在CustomerDWD 单例对象Main方法中调用save方法save(customerDetailDF, OfflineTableDefine.customerDetail)3.5、删除缓存数据为了释放资源，客户明细宽表数据计算完成以后，需要将缓存的源表数据删除。//移除缓存 customerDetailDF.unpersist codesDF.unpersist expressPackageDF.unpersist customerSenderDF.unpersist customerDF.unpersist3.6、完整代码package cn.it.logistics.offline.dwd import cn.it.logistics.common.{CodeTypeMapping, Configuration, OfflineTableDefine, SparkUtils, TableMapping} import cn.it.logistics.offline.OfflineApp import org.apache.spark.SparkConf import org.apache.spark.sql.{DataFrame, SparkSession} import org.apache.spark.storage.StorageLevel import org.apache.spark.sql.functions._ import org.apache.spark.sql.types.IntegerType /** * 客户主题数据的拉宽操作 */ object CustomerDWD extends OfflineApp { //定义应用的名称 val appName = this.getClass.getSimpleName def main(args: Array[String]): Unit = { /** * 实现步骤： * 1）初始化sparkConf对象 * 2）创建sparkSession对象 * 3）加载kudu中的事实表和维度表的数据（将加载后的数据进行缓存） * 4）定义维度表与事实表的关联 * 5）将拉宽后的数据再次写回到kudu数据库中（DWD明细层） * 5.1：创建车辆明细宽表的schema表结构 * 5.2：创建车辆宽表（判断宽表是否存在，如果不存在则创建） * 5.3：将数据写入到kudu中 * 6）将缓存的数据删除掉 * 7）停止任务 */ //1）初始化sparkConf对象 val sparkConf: SparkConf = SparkUtils.autoSettingEnv( SparkUtils.sparkConf(appName) ) //2）创建sparkSession对象 val sparkSession: SparkSession = SparkUtils.getSparkSession(sparkConf) sparkSession.sparkContext.setLogLevel(Configuration.LOG_OFF) //数据处理 execute(sparkSession) } /** * 数据处理 * * @param sparkSession */ override def execute(sparkSession: SparkSession): Unit = { //导入隐士转换 import sparkSession.implicits._ val customerSenderInfoDF: DataFrame = getKuduSource(sparkSession, TableMapping.consumerSenderInfo, Configuration.isFirstRunnable).persist(StorageLevel.DISK_ONLY_2) val customerDF = getKuduSource(sparkSession, TableMapping.customer, true).persist(StorageLevel.DISK_ONLY_2) val expressPageageDF = getKuduSource(sparkSession, TableMapping.expressPackage, true).persist(StorageLevel.DISK_ONLY_2) val codesDF: DataFrame = getKuduSource(sparkSession, TableMapping.codes, true).persist(StorageLevel.DISK_ONLY_2) val customerTypeDF = codesDF.where($"type" === CodeTypeMapping.CustomType) //TODO 4）定义维度表与事实表的关联关系 val left_outer = "left_outer" /** * 获取每个用户的首尾单发货信息及发货件数和总金额 */ val customerSenderDetailInfoDF: DataFrame = customerSenderInfoDF.join(expressPageageDF, expressPageageDF("id") === customerSenderInfoDF("pkgId"), left_outer) .groupBy(customerSenderInfoDF("ciid")) .agg(min(customerSenderInfoDF("id")).alias("first_id"), max(customerSenderInfoDF("id")).alias("last_id"), min(expressPageageDF("cdt")).alias("first_cdt"), max(expressPageageDF("cdt")).alias("last_cdt"), count(customerSenderInfoDF("id")).alias("totalCount"), sum(expressPageageDF("actualAmount")).alias("totalAmount") ) val customerDetailDF: DataFrame = customerDF .join(customerSenderDetailInfoDF, customerDF("id") === customerSenderInfoDF("ciid"), left_outer) .join(customerTypeDF, customerDF("type") === customerTypeDF("code").cast(IntegerType), left_outer) .sort(customerDF("cdt").asc) .select( customerDF("id"), customerDF("name"), customerDF("tel"), customerDF("mobile"), customerDF("type").cast(IntegerType), customerTypeDF("codeDesc").as("type_name"), customerDF("isownreg").as("is_own_reg"), customerDF("regdt").as("regdt"), customerDF("regchannelid").as("reg_channel_id"), customerDF("state"), customerDF("cdt"), customerDF("udt"), customerDF("lastlogindt").as("last_login_dt"), customerDF("remark"), customerSenderDetailInfoDF("first_id").as("first_sender_id"), //首次寄件id customerSenderDetailInfoDF("last_id").as("last_sender_id"), //尾次寄件id customerSenderDetailInfoDF("first_cdt").as("first_sender_cdt"), //首次寄件时间 customerSenderDetailInfoDF("last_cdt").as("last_sender_cdt"), //尾次寄件时间 customerSenderDetailInfoDF("totalCount"), //寄件总次数 customerSenderDetailInfoDF("totalAmount") //总金额 ) save(customerDetailDF, OfflineTableDefine.customerDetail) // 5.4：将缓存的数据删除掉 customerDF.unpersist() customerSenderInfoDF.unpersist() expressPageageDF.unpersist() customerTypeDF.unpersist() sparkSession.stop() } }

文章

SQL · 搜索推荐 · 大数据 · 缓存 · 调度 · 分布式计算 · Spark

2022-05-19

客快物流大数据项目(六十六)：车辆主题（下）

2、SQL语句2.1、拉宽网点车辆表SELECT ttl."id" , ttl."brand" , ttl."model" , ttl."type" , TTL ."given_load" , TTL ."load_cn_unit" , TTL ."load_en_unit" , TTL ."buy_dt" , TTL ."license_plate" , ttl."state" , ttl."cdt" , ttl."udt" , ttl."remark" , dot."id" AS dot_id, dot."dot_name" , dot."dot_number" , dot."dot_addr" , dot."dot_gis_addr" , dot."dot_tel" , dot."manage_area_id" , dot."manage_area_gis" , COMPANY ."id" AS company_id, COMPANY ."company_name", COMPANY ."company_number", COMPANY ."city_id", COMPANY ."company_addr", COMPANY ."company_addr_gis", COMPANY ."company_tel", COMPANY ."is_sub_company" FROM "tbl_transport_tool" ttl LEFT JOIN "tbl_dot_transport_tool" tdtl ON ttl."id" = tdtl."transport_tool_id" LEFT JOIN "tbl_dot" dot ON DOT ."id" = TDTL ."dot_id" LEFT JOIN "tbl_company_dot_map" companydot ON companydot."dot_id" = TDTL ."dot_id" LEFT JOIN "tbl_company" company ON company."id" = companydot."company_id"2.2、拉宽仓库车辆表SELECT ttl."id" , ttl."brand" , ttl."model" , ttl."type" , ttl."given_load" , ttl."load_cn_unit" , ttl."load_en_unit" , ttl."buy_dt" , ttl."license_plate" , ttl."state" , ttl."cdt" , ttl."udt" , ttl."remark" , warehouse."id" , warehouse."name", warehouse."addr", warehouse."addr_gis", warehouse."employee_id", warehouse."type", warehouse."area", warehouse."is_lease", COMPANY ."id" AS company_id, COMPANY ."company_name", COMPANY ."company_number", COMPANY ."city_id", COMPANY ."company_addr", COMPANY ."company_addr_gis", COMPANY ."company_tel", COMPANY ."is_sub_company" FROM "tbl_warehouse_transport_tool" twt LEFT JOIN "tbl_transport_tool" ttl ON twt."transport_tool_id" = ttl."id" LEFT JOIN "tbl_warehouse" warehouse ON WAREHOUSE ."id" = twt."warehouse_id" LEFT JOIN "tbl_company_warehouse_map" warehouse_map ON warehouse_map."warehouse_id" = warehouse."id" LEFT JOIN "tbl_company" company ON company."id" = warehouse_map."company_id"3、Spark实现实现步骤：在dwd目录下创建 TransportToolDWD 单例对象，继承自OfflineApp特质初始化环境的参数，创建SparkSession对象获取运输工具表（tbl_transport_tool）数据，并缓存数据判断是否是首次运行，如果是首次运行的话，则全量装载数据（含历史数据）获取网点运输工具关联表（tbl_dot_transport_tool）数据，并缓存数据获取网点表（tbl_dot）数据，并缓存数据获取公司网点关联表（tbl_company_dot_map）数据，并缓存数据获取仓库运输工具关联表（tbl_warehouse_transport_tool）数据，并缓存数据获取公司仓库关联表（tbl_company_warehouse_map）数据，并缓存数据获取仓库表（tbl_warehouse）数据，并缓存数据获取公司表（tbl_company）数据，并缓存数据根据以下方式拉宽仓库车辆明细数据根据交通工具id，在交通工具表中获取交通工具数据根据网点id，在网点表中获取网点数据根据公司id，在公司表中获取公司数据根据仓库id，在仓库表中获取仓库数据创建网点车辆明细宽表（若存在则不创建）创建仓库车辆明细宽表（若存在则不创建）将仓库车辆明细宽表数据写入到kudu数据表中删除缓存数据3.1、初始化环境变量初始化运单明细拉宽作业的环境变量package cn.it.logistics.offline.dwd import cn.it.logistics.common.{Configuration, SparkUtils} import cn.it.logistics.offline.OfflineApp import org.apache.spark.SparkConf import org.apache.spark.sql.SparkSession /** * 车辆主题开发 * 将车辆相关的表进行关联拉宽，将拉宽后的数据写入到车辆宽表中 * 1）网点车辆关联表->派送网点所拥有的车辆（三轮车） * 2）仓库车辆关联表->仓库与仓库之间的运输工具（货车、货机） */ object TransportToolDWD extends OfflineApp{ //定义应用的名称 val appName = this.getClass.getSimpleName /** * 入口函数 * @param args */ def main(args: Array[String]): Unit = { /** * 实现步骤： * 1）初始化sparkConf对象 * 2）创建sparkSession对象 * 3）加载kudu中的事实表和维度表的数据（将加载后的数据进行缓存） * 4）定义维度表与事实表的关联 * 5）将拉宽后的数据再次写回到kudu数据库中（DWD明细层） * 5.1：创建车辆明细宽表的schema表结构 * 5.2：创建车辆宽表（判断宽表是否存在，如果不存在则创建） * 5.3：将数据写入到kudu中 * 6）将缓存的数据删除掉 * 7）停止任务 */ //1）初始化sparkConf对象 val sparkConf: SparkConf = SparkUtils.autoSettingEnv( SparkUtils.sparkConf(appName) ) //2）创建sparkSession对象 val sparkSession: SparkSession = SparkUtils.getSparkSession(sparkConf) sparkSession.sparkContext.setLogLevel(Configuration.LOG_OFF) //数据处理 execute(sparkSession) } /** * 数据处理 * * @param sparkSession */ override def execute(sparkSession: SparkSession): Unit = { sparkSession.stop() } } 3.2、加载运输工具表及车辆相关的表并缓存加载运输工具表的时候，需要指定日期条件，因为车辆主题最终需要Azkaban定时调度执行，每天执行一次增量数据，因此需要指定日期。判断是否是首次运行，如果是首次运行的话，则全量装载数据（含历史数据）//TODO 3）加载kudu中的事实表和维度表的数据（将加载后的数据进行缓存） //加载车辆表数据（事实表） val ttDF: DataFrame = getKuduSource(sparkSession, TableMapping.transportTool, Configuration.isFirstRunnable).persist(StorageLevel.DISK_ONLY_2) //加载仓库车辆表数据 val ttDotDF: DataFrame = getKuduSource(sparkSession, TableMapping.dotTransportTool, true).persist(StorageLevel.DISK_ONLY_2) //加载网点表的数据 val dotDF: DataFrame = getKuduSource(sparkSession, TableMapping.dot, true).persist(StorageLevel.DISK_ONLY_2) //加载公司网点关联表的数据 val companyDotDF: DataFrame = getKuduSource(sparkSession, TableMapping.companyDotMap, true).persist(StorageLevel.DISK_ONLY_2) //加载公司表的数据 val companyDF: DataFrame = getKuduSource(sparkSession, TableMapping.company, true).persist(StorageLevel.DISK_ONLY_2) //加载仓库车辆关联表数据（事实表） val ttWsDF: DataFrame = getKuduSource(sparkSession, TableMapping.warehouseTransportTool, Configuration.isFirstRunnable).persist(StorageLevel.DISK_ONLY_2) //加载仓库公司关联表 val companyWareHouseMapDF: DataFrame = getKuduSource(sparkSession, TableMapping.companyWarehouseMap, true).persist(StorageLevel.DISK_ONLY_2) //加载仓库表数据 val wsDF: DataFrame = getKuduSource(sparkSession, TableMapping.warehouse, true).persist(StorageLevel.DISK_ONLY_2) //加载物流码表数据 val codesDF: DataFrame = getKuduSource(sparkSession, TableMapping.codes, true).persist(StorageLevel.DISK_ONLY_2) import sparkSession.implicits._ //获取运输工具类型 val transportTypeDF: DataFrame = codesDF.where($"type" === CodeTypeMapping.TransportType).select($"code".as("ttType"), $"codeDesc".as("ttTypeName")) //获取运输工具状态 val transportStatusDF: DataFrame = codesDF.where($"type" === CodeTypeMapping.TransportStatus).select($"code".as("ttStatus"), $"codeDesc".as("ttStateName"))3.3、定义网点车辆宽表的关联关系为了在DWS层任务中方便的获取每日增量网点车辆表数据（根据日期），因此在DataFrame基础上动态增加列（day），指定日期格式为yyyyMMdd代码如下：//TODO 4）定义维度表与事实表的关联 val left_outer: String = "left_outer" // 4.1：拉宽网点车辆表 val ttDotDetailDF = ttDotDF.join(ttDF, ttDotDF.col("transportToolId") === ttDF.col("id"), left_outer) //网点车辆表关联车辆表 .join(transportTypeDF, transportTypeDF("ttType") === ttDF("type"), left_outer) //车辆表类型关联字典表类型 .join(transportStatusDF, transportStatusDF("ttStatus") === ttDF("state"), left_outer) //车辆表状态管理字典表状态 .join(dotDF, dotDF.col("id") === ttDotDF.col("dotId"), left_outer) //网点车辆表关联网点 .join(companyDotDF, ttDotDF.col("dotId") === companyDotDF.col("dotId"), left_outer) //网点车辆管连网点公司关联表 .join(companyDF, companyDotDF.col("companyId") === companyDF.col("id"), left_outer) //网点车辆表关联公司表 .withColumn("day", date_format(ttDotDF("cdt"), "yyyyMMdd"))//虚拟列,可以根据这个日期列作为分区字段，可以保证同一天的数据保存在同一个分区中 .sort(ttDF.col("cdt").asc) .select( ttDF("id"), //车辆表id ttDF("brand"), //车辆表brand ttDF("model"), //车辆表model ttDF("type").cast(IntegerType), //车辆表type transportTypeDF("ttTypeName").as("type_name"), // 车辆表type对应字典表车辆类型的具体描述 ttDF("givenLoad").cast(IntegerType).as("given_load"), //车辆表given_load ttDF("loadCnUnit").as("load_cn_unit"), //车辆表load_cn_unit ttDF("loadEnUnit").as("load_en_unit"), //车辆表load_en_unit ttDF("buyDt").as("buy_dt"), //车辆表buy_dt ttDF("licensePlate").as("license_plate"), //车辆表license_plate ttDF("state").cast(IntegerType), //车辆表state transportStatusDF("ttStateName").as("state_name"), // 车辆表state对应字典表类型的具体描述 ttDF("cdt"), //车辆表cdt ttDF("udt"), //车辆表udt ttDF("remark"), //车辆表remark dotDF("id").as("dot_id"), //网点表dot_id dotDF("dotNumber").as("dot_number"), //网点表dot_number dotDF("dotName").as("dot_name"), //网点表dot_name dotDF("dotAddr").as("dot_addr"), //网点表dot_addr dotDF("dotGisAddr").as("dot_gis_addr"), //网点表dot_gis_addr dotDF("dotTel").as("dot_tel"), //网点表dot_tel dotDF("manageAreaId").as("manage_area_id"), //网点表manage_area_id dotDF("manageAreaGis").as("manage_area_gis"), //网点表manage_area_gis companyDF("id").alias("company_id"), //公司表id companyDF("companyName").as("company_name"), //公司表company_name companyDF("cityId").as("city_id"), //公司表city_id companyDF("companyNumber").as("company_number"), //公司表company_number companyDF("companyAddr").as("company_addr"), //公司表company_addr companyDF("companyAddrGis").as("company_addr_gis"), //公司表company_addr_gis companyDF("companyTel").as("company_tel"), //公司表company_tel companyDF("isSubCompany").as("is_sub_company"), //公司表is_sub_company $"day" )3.4、创建网点车辆明细宽表并将网点车辆明细数据写入到kudu数据表中网点车辆明细宽表数据需要保存到kudu中，因此在第一次执行网点车辆明细拉宽操作时，网点车辆明细宽表是不存在的，因此需要实现自动判断宽表是否存在，如果不存在则创建实现步骤：在TransportToolDWD 单例对象中调用save方法实现过程：在TransportToolDWD 单例对象Main方法中调用save方法//TODO 5）将拉宽后的数据再次写回到kudu数据库中（DWD明细层） save(ttDotDetailDF, OfflineTableDefine.dotTransportToolDetail)3.5、定义仓库车辆宽表的关联关系为了在DWS层任务中方便的获取每日增量仓库车辆表数据（根据日期），因此在DataFrame基础上动态增加列（day），指定日期格式为yyyyMMdd代码如下：// 4.2：拉宽仓库车辆表 // 拉宽仓库车辆表 val ttWsDetailDF = ttWsDF.join(ttDF, ttWsDF.col("transportToolId") === ttDF.col("id"), left_outer) //仓库车辆表关联车辆表 .join(transportTypeDF, transportTypeDF("ttType") === ttDF("type"), left_outer) //车辆表类型关联字典表类型 .join(transportStatusDF, transportStatusDF("ttStatus") === ttDF("state"), left_outer) //车辆表状态管理字典表状态 .join(wsDF, wsDF.col("id") === ttWsDF.col("warehouseId"), left_outer) //仓库车辆表关联仓库 .join(companyWareHouseMapDF, ttWsDF.col("warehouseId") === companyWareHouseMapDF.col("warehouseId"), left_outer) //仓库车辆管连仓库公司关联表 .join(companyDF, companyDF.col("id") === companyWareHouseMapDF.col("companyId"), left_outer) .withColumn("day", date_format(ttWsDF("cdt"), "yyyyMMdd"))//虚拟列,可以根据这个日期列作为分区字段，可以保证同一天的数据保存在同一个分区中 .sort(ttDF.col("cdt").asc) .select( ttDF("id"), //车辆表id ttDF("brand"), //车辆表brand ttDF("model"), //车辆表model ttDF("type").cast(IntegerType), //车辆表type transportTypeDF("ttTypeName").as("type_name"), // 车辆表type对应字典表车辆类型的具体描述 ttDF("givenLoad").as("given_load").cast(IntegerType), //车辆表given_load ttDF("loadCnUnit").as("load_cn_unit"), //车辆表load_cn_unit ttDF("loadEnUnit").as("load_en_unit"), //车辆表load_en_unit ttDF("buyDt").as("buy_dt"), //车辆表buy_dt ttDF("licensePlate").as("license_plate"), //车辆表license_plate ttDF("state").cast(IntegerType), //车辆表state transportStatusDF("ttStateName").as("state_name"), // 车辆表type对应字典表车辆类型的具体描述 ttDF("cdt"), //车辆表cdt ttDF("udt"), //车辆表udt ttDF("remark"), //车辆表remark wsDF("id").as("ws_id"), //仓库表id wsDF("name"), //仓库表name wsDF("addr"), //仓库表addr wsDF("addrGis").as("addr_gis"), //仓库表addr_gis wsDF("employeeId").as("employee_id"), //仓库表employee_id wsDF("type").as("ws_type").cast(IntegerType), //仓库表type wsDF("area"), //仓库表area wsDF("isLease").as("is_lease").cast(IntegerType), //仓库表is_lease companyDF("id").alias("company_id"), //公司表id companyDF("companyName").as("company_name"), //公司表company_name companyDF("cityId").as("city_id"), //公司表city_id companyDF("companyNumber").as("company_number"), //公司表company_number companyDF("companyAddr").as("company_addr"), //公司表company_addr companyDF("companyAddrGis").as("company_addr_gis"), //公司表company_addr_gis companyDF("companyTel").as("company_tel"), //公司表company_tel companyDF("isSubCompany").as("is_sub_company"), //公司表is_sub_company $"day" )3.6、创建仓库车辆明细宽表并将仓库车辆明细数据写入到kudu数据表中仓库车辆明细宽表数据需要保存到kudu中，因此在第一次执行仓库车辆明细拉宽操作时，仓库车辆明细宽表是不存在的，因此需要实现自动判断宽表是否存在，如果不存在则创建实现步骤：在TransportToolDWD 单例对象中调用save方法实现过程：在TransportToolDWD 单例对象Main方法中调用save方法save(ttWsDetailDF, OfflineTableDefine.warehouseTransportToolDetail)3.7、删除缓存数据为了释放资源，车辆明细宽表数据计算完成以后，需要将缓存的源表数据删除。//TODO 6）将缓存的数据删除掉 ttDF.unpersist() ttDotDF.unpersist() dotDF.unpersist() companyDotDF.unpersist() companyDF.unpersist() ttWsDF.unpersist() companyWareHouseMapDF.unpersist() wsDF.unpersist() codesDF.unpersist()3.8、完整代码package cn.it.logistics.offline.dwd import cn.it.logistics.common.{CodeTypeMapping, Configuration, OfflineTableDefine, SparkUtils, TableMapping} import cn.it.logistics.offline.OfflineApp import org.apache.spark.SparkConf import org.apache.spark.sql.functions.date_format import org.apache.spark.sql.types.IntegerType import org.apache.spark.sql.{DataFrame, SparkSession} import org.apache.spark.storage.StorageLevel /** * 车辆主题开发 * 将车辆相关的表进行关联拉宽，将拉宽后的数据写入到车辆宽表中 * 1）网点车辆关联表->派送网点所拥有的车辆（三轮车） * 2）仓库车辆关联表->仓库与仓库之间的运输工具（货车、货机） */ object TransportToolDWD extends OfflineApp{ //定义应用的名称 val appName = this.getClass.getSimpleName /** * 入口函数 * @param args */ def main(args: Array[String]): Unit = { /** * 实现步骤： * 1）初始化sparkConf对象 * 2）创建sparkSession对象 * 3）加载kudu中的事实表和维度表的数据（将加载后的数据进行缓存） * 4）定义维度表与事实表的关联 * 5）将拉宽后的数据再次写回到kudu数据库中（DWD明细层） * 5.1：创建车辆明细宽表的schema表结构 * 5.2：创建车辆宽表（判断宽表是否存在，如果不存在则创建） * 5.3：将数据写入到kudu中 * 6）将缓存的数据删除掉 * 7）停止任务 */ //1）初始化sparkConf对象 val sparkConf: SparkConf = SparkUtils.autoSettingEnv( SparkUtils.sparkConf(appName) ) //2）创建sparkSession对象 val sparkSession: SparkSession = SparkUtils.getSparkSession(sparkConf) sparkSession.sparkContext.setLogLevel(Configuration.LOG_OFF) //数据处理 execute(sparkSession) } /** * 数据处理 * * @param sparkSession */ override def execute(sparkSession: SparkSession): Unit = { //TODO 3）加载kudu中的事实表和维度表的数据（将加载后的数据进行缓存） //加载车辆表数据（事实表） val ttDF: DataFrame = getKuduSource(sparkSession, TableMapping.transportTool, Configuration.isFirstRunnable).persist(StorageLevel.DISK_ONLY_2) //加载仓库车辆表数据 val ttDotDF: DataFrame = getKuduSource(sparkSession, TableMapping.dotTransportTool, true).persist(StorageLevel.DISK_ONLY_2) //加载网点表的数据 val dotDF: DataFrame = getKuduSource(sparkSession, TableMapping.dot, true).persist(StorageLevel.DISK_ONLY_2) //加载公司网点关联表的数据 val companyDotDF: DataFrame = getKuduSource(sparkSession, TableMapping.companyDotMap, true).persist(StorageLevel.DISK_ONLY_2) //加载公司表的数据 val companyDF: DataFrame = getKuduSource(sparkSession, TableMapping.company, true).persist(StorageLevel.DISK_ONLY_2) //加载仓库车辆关联表数据（事实表） val ttWsDF: DataFrame = getKuduSource(sparkSession, TableMapping.warehouseTransportTool, Configuration.isFirstRunnable).persist(StorageLevel.DISK_ONLY_2) //加载仓库公司关联表 val companyWareHouseMapDF: DataFrame = getKuduSource(sparkSession, TableMapping.companyWarehouseMap, true).persist(StorageLevel.DISK_ONLY_2) //加载仓库表数据 val wsDF: DataFrame = getKuduSource(sparkSession, TableMapping.warehouse, true).persist(StorageLevel.DISK_ONLY_2) //加载物流码表数据 val codesDF: DataFrame = getKuduSource(sparkSession, TableMapping.codes, true).persist(StorageLevel.DISK_ONLY_2) import sparkSession.implicits._ //获取运输工具类型 val transportTypeDF: DataFrame = codesDF.where($"type" === CodeTypeMapping.TransportType).select($"code".as("ttType"), $"codeDesc".as("ttTypeName")) //获取运输工具状态 val transportStatusDF: DataFrame = codesDF.where($"type" === CodeTypeMapping.TransportStatus).select($"code".as("ttStatus"), $"codeDesc".as("ttStateName")) //TODO 4）定义维度表与事实表的关联 val left_outer: String = "left_outer" // 4.1：拉宽网点车辆表 val ttDotDetailDF = ttDotDF.join(ttDF, ttDotDF.col("transportToolId") === ttDF.col("id"), left_outer) //网点车辆表关联车辆表 .join(transportTypeDF, transportTypeDF("ttType") === ttDF("type"), left_outer) //车辆表类型关联字典表类型 .join(transportStatusDF, transportStatusDF("ttStatus") === ttDF("state"), left_outer) //车辆表状态管理字典表状态 .join(dotDF, dotDF.col("id") === ttDotDF.col("dotId"), left_outer) //网点车辆表关联网点 .join(companyDotDF, ttDotDF.col("dotId") === companyDotDF.col("dotId"), left_outer) //网点车辆管连网点公司关联表 .join(companyDF, companyDotDF.col("companyId") === companyDF.col("id"), left_outer) //网点车辆表关联公司表 .withColumn("day", date_format(ttDotDF("cdt"), "yyyyMMdd"))//虚拟列,可以根据这个日期列作为分区字段，可以保证同一天的数据保存在同一个分区中 .sort(ttDF.col("cdt").asc) .select( ttDF("id"), //车辆表id ttDF("brand"), //车辆表brand ttDF("model"), //车辆表model ttDF("type").cast(IntegerType), //车辆表type transportTypeDF("ttTypeName").as("type_name"), // 车辆表type对应字典表车辆类型的具体描述 ttDF("givenLoad").cast(IntegerType).as("given_load"), //车辆表given_load ttDF("loadCnUnit").as("load_cn_unit"), //车辆表load_cn_unit ttDF("loadEnUnit").as("load_en_unit"), //车辆表load_en_unit ttDF("buyDt").as("buy_dt"), //车辆表buy_dt ttDF("licensePlate").as("license_plate"), //车辆表license_plate ttDF("state").cast(IntegerType), //车辆表state transportStatusDF("ttStateName").as("state_name"), // 车辆表state对应字典表类型的具体描述 ttDF("cdt"), //车辆表cdt ttDF("udt"), //车辆表udt ttDF("remark"), //车辆表remark dotDF("id").as("dot_id"), //网点表dot_id dotDF("dotNumber").as("dot_number"), //网点表dot_number dotDF("dotName").as("dot_name"), //网点表dot_name dotDF("dotAddr").as("dot_addr"), //网点表dot_addr dotDF("dotGisAddr").as("dot_gis_addr"), //网点表dot_gis_addr dotDF("dotTel").as("dot_tel"), //网点表dot_tel dotDF("manageAreaId").as("manage_area_id"), //网点表manage_area_id dotDF("manageAreaGis").as("manage_area_gis"), //网点表manage_area_gis companyDF("id").alias("company_id"), //公司表id companyDF("companyName").as("company_name"), //公司表company_name companyDF("cityId").as("city_id"), //公司表city_id companyDF("companyNumber").as("company_number"), //公司表company_number companyDF("companyAddr").as("company_addr"), //公司表company_addr companyDF("companyAddrGis").as("company_addr_gis"), //公司表company_addr_gis companyDF("companyTel").as("company_tel"), //公司表company_tel companyDF("isSubCompany").as("is_sub_company"), //公司表is_sub_company $"day" ) // 4.2：拉宽仓库车辆表 // 拉宽仓库车辆表 val ttWsDetailDF = ttWsDF.join(ttDF, ttWsDF.col("transportToolId") === ttDF.col("id"), left_outer) //仓库车辆表关联车辆表 .join(transportTypeDF, transportTypeDF("ttType") === ttDF("type"), left_outer) //车辆表类型关联字典表类型 .join(transportStatusDF, transportStatusDF("ttStatus") === ttDF("state"), left_outer) //车辆表状态管理字典表状态 .join(wsDF, wsDF.col("id") === ttWsDF.col("warehouseId"), left_outer) //仓库车辆表关联仓库 .join(companyWareHouseMapDF, ttWsDF.col("warehouseId") === companyWareHouseMapDF.col("warehouseId"), left_outer) //仓库车辆管连仓库公司关联表 .join(companyDF, companyDF.col("id") === companyWareHouseMapDF.col("companyId"), left_outer) .withColumn("day", date_format(ttWsDF("cdt"), "yyyyMMdd"))//虚拟列,可以根据这个日期列作为分区字段，可以保证同一天的数据保存在同一个分区中 .sort(ttDF.col("cdt").asc) .select( ttDF("id"), //车辆表id ttDF("brand"), //车辆表brand ttDF("model"), //车辆表model ttDF("type").cast(IntegerType), //车辆表type transportTypeDF("ttTypeName").as("type_name"), // 车辆表type对应字典表车辆类型的具体描述 ttDF("givenLoad").as("given_load").cast(IntegerType), //车辆表given_load ttDF("loadCnUnit").as("load_cn_unit"), //车辆表load_cn_unit ttDF("loadEnUnit").as("load_en_unit"), //车辆表load_en_unit ttDF("buyDt").as("buy_dt"), //车辆表buy_dt ttDF("licensePlate").as("license_plate"), //车辆表license_plate ttDF("state").cast(IntegerType), //车辆表state transportStatusDF("ttStateName").as("state_name"), // 车辆表type对应字典表车辆类型的具体描述 ttDF("cdt"), //车辆表cdt ttDF("udt"), //车辆表udt ttDF("remark"), //车辆表remark wsDF("id").as("ws_id"), //仓库表id wsDF("name"), //仓库表name wsDF("addr"), //仓库表addr wsDF("addrGis").as("addr_gis"), //仓库表addr_gis wsDF("employeeId").as("employee_id"), //仓库表employee_id wsDF("type").as("ws_type").cast(IntegerType), //仓库表type wsDF("area"), //仓库表area wsDF("isLease").as("is_lease").cast(IntegerType), //仓库表is_lease companyDF("id").alias("company_id"), //公司表id companyDF("companyName").as("company_name"), //公司表company_name companyDF("cityId").as("city_id"), //公司表city_id companyDF("companyNumber").as("company_number"), //公司表company_number companyDF("companyAddr").as("company_addr"), //公司表company_addr companyDF("companyAddrGis").as("company_addr_gis"), //公司表company_addr_gis companyDF("companyTel").as("company_tel"), //公司表company_tel companyDF("isSubCompany").as("is_sub_company"), //公司表is_sub_company $"day" ) //TODO 5）将拉宽后的数据再次写回到kudu数据库中（DWD明细层） save(ttDotDetailDF, OfflineTableDefine.dotTransportToolDetail) save(ttWsDetailDF, OfflineTableDefine.warehouseTransportToolDetail) //TODO 6）将缓存的数据删除掉 ttDF.unpersist() ttDotDF.unpersist() dotDF.unpersist() companyDotDF.unpersist() companyDF.unpersist() ttWsDF.unpersist() companyWareHouseMapDF.unpersist() wsDF.unpersist() codesDF.unpersist() sparkSession.stop() } }五、车辆数据指标开发1、计算的字段2、Spark实现实现步骤：在dws目录下创建 TransportToolDWS 单例对象，继承自OfflineApp特质初始化环境的参数，创建SparkSession对象根据指定的日期获取拉宽后的车辆主题宽表（tbl_dot_transport_tool_detail、tbl_dot_transport_tool_detail）增量数据，并缓存数据判断是否是首次运行，如果是首次运行的话，则全量装载数据（含历史数据）指标计算各网点发车次数各网点最大发车次数各网点最小发车次数各网点平均发车次数各区域发车次数各区域最大发车次数各区域最小发车次数各区域平均发车次数各公司发车次数各公司最大发车次数各公司最小发车次数各公司平均发车次数获取当前时间yyyyMMddHH构建要持久化的指标数据（需要判断计算的指标是否有值，若没有需要赋值默认值）通过StructType构建指定Schema创建车辆主题指标数据表（若存在则不创建）持久化指标数据到kudu表2.1、初始化环境变量package cn.it.logistics.offline.dws import cn.it.logistics.common.{Configuration, SparkUtils} import cn.it.logistics.offline.OfflineApp import org.apache.spark.SparkConf import org.apache.spark.sql.SparkSession /** * 车辆主题指标开发 */ object TransportToolDWS extends OfflineApp{ //定义应用程序的名称 val appName = this.getClass.getSimpleName /** * 入口函数 * @param args */ def main(args: Array[String]): Unit = { /** * 实现步骤： * 1）创建SparkConf对象 * 2）创建SparkSession对象 * 3）读取车辆明细宽表的数据 * 4）对车辆明细宽表的数据进行指标的计算 * 5）将计算好的指标数据写入到kudu数据库中 * 5.1：定义指标结果表的schema信息 * 5.2：组织需要写入到kudu表的数据 * 5.3：判断指标结果表是否存在，如果不存在则创建 * 5.4：将数据写入到kudu表中 * 6）删除缓存数据 * 7）停止任务，退出sparksession */ //TODO 1）创建SparkConf对象 val sparkConf: SparkConf = SparkUtils.autoSettingEnv( SparkUtils.sparkConf(appName) ) //TODO 2）创建SparkSession对象 val sparkSession: SparkSession = SparkUtils.getSparkSession(sparkConf) sparkSession.sparkContext.setLogLevel(Configuration.LOG_OFF) //处理数据 execute(sparkSession) } /** * 数据处理 * * @param sparkSession */ override def execute(sparkSession: SparkSession): Unit = { sparkSession.stop() } }2.2、加载车辆宽表增量数据并缓存加载车辆宽表的时候，需要指定日期条件，因为车辆主题最终需要Azkaban定时调度执行，每天执行一次增量数据，因此需要指定日期。//TODO 3）读取车辆明细宽表的数据 val ttDotDetailDF: DataFrame = getKuduSource(sparkSession, OfflineTableDefine.dotTransportToolDetail, Configuration.isFirstRunnable).persist(StorageLevel.DISK_ONLY_2) val ttWarehouseDetailDF: DataFrame = getKuduSource(sparkSession, OfflineTableDefine.warehouseTransportToolDetail, Configuration.isFirstRunnable).persist(StorageLevel.DISK_ONLY_2)2.3、指标计算//根据网点车辆的日期进行分组 val ttDotDetailGroupByDayDF: DataFrame = ttDotDetailDF.select("day").groupBy("day").count().cache() //导入隐式转换 import sparkSession.implicits._ //定义计算好的指标结果集合对象 val dotRows: ArrayBuffer[Row] = ArrayBuffer[Row]() //循环遍历网点车辆每个日期的车辆明细宽表数据 ttDotDetailGroupByDayDF.collect().foreach(row=> { //获取到要处理的数据所在的日期 val day: String = row.getAs[String](0) //返回指定日期的仓库明细数据 val ttDotDetailByDayDF: DataFrame = ttDotDetailDF.where(col("day") === day).toDF().persist(StorageLevel.DISK_ONLY_2) //各网点的发车次数（西三旗：10，西二旗：20） val ttDotTotalCountDF: DataFrame = ttDotDetailByDayDF.groupBy($"dot_id").agg(count("dot_id").alias("cnt")).persist(StorageLevel.DISK_ONLY_2) //计算各网点的总发车次数 val ttDotTotalCount: Row = ttDotTotalCountDF.agg(sum("cnt")).first() //各网点的最大发车次数、最小发车次数、平均发车次数 val maxAndMinAndAvgTTDotTotalCount: Row = ttDotTotalCountDF.agg(max("cnt"), min("cnt"), avg("cnt")).first() val maxTTDotTotalCount: Any = maxAndMinAndAvgTTDotTotalCount(0) val minTTDotTotalCount: Any = maxAndMinAndAvgTTDotTotalCount(1) val avgTTDotTotalCount: Any = maxAndMinAndAvgTTDotTotalCount(2) // 各区域发车次数 val areaDotTotalCountDF = ttDotDetailByDayDF.groupBy("city_id").agg(count($"id").alias("cnt")).persist(StorageLevel.DISK_ONLY_2) val areaDotTotalCount = areaDotTotalCountDF.agg(sum($"cnt")).first() // 各区域最大发车次数、最小发车次数和平均发车次数 val maxAndMinAndAvgAreaDotTotalCount = areaDotTotalCountDF.agg(max($"cnt"), min($"cnt"), avg($"cnt")).first() val maxAreaDotTotalCount = maxAndMinAndAvgAreaDotTotalCount(0) val minAreaDotTotalCount = maxAndMinAndAvgAreaDotTotalCount(1) val avgAreaDotTotalCount = maxAndMinAndAvgAreaDotTotalCount(2) // 各公司发车次数 val companyDotTotalCountDF = ttDotDetailByDayDF.groupBy("company_id").agg(count($"id").alias("cnt")).persist(StorageLevel.DISK_ONLY_2) val companyDotTotalCount = companyDotTotalCountDF.agg(sum($"cnt")).first() // 各公司最大发车次数、最小发车次数和平均发车次数 val maxAndMinAndAvgCompanyDotTotalCount = companyDotTotalCountDF.agg(max($"cnt"), min($"cnt"), avg($"cnt")).first() val maxCompanyDotTotalCount = maxAndMinAndAvgCompanyDotTotalCount(0) val minCompanyDotTotalCount = maxAndMinAndAvgCompanyDotTotalCount(1) val avgCompanyDotTotalCount = maxAndMinAndAvgCompanyDotTotalCount(2) //将计算好的指标数据封装到row对象中 val rowInfo: Row = Row( day, if(ttDotTotalCount.isNullAt(0)) 0L else ttDotTotalCount.get(0).asInstanceOf[Number].longValue(), if(maxTTDotTotalCount==null) 0L else maxTTDotTotalCount.asInstanceOf[Number].longValue(), if(minTTDotTotalCount==null) 0L else minTTDotTotalCount.asInstanceOf[Number].longValue(), if(avgTTDotTotalCount==null) 0L else avgTTDotTotalCount.asInstanceOf[Number].longValue(), if(areaDotTotalCount.isNullAt(0)) 0L else areaDotTotalCount.get(0).asInstanceOf[Number].longValue(), if(maxAreaDotTotalCount==null) 0L else maxAreaDotTotalCount.asInstanceOf[Number].longValue(), if(minAreaDotTotalCount==null) 0L else minAreaDotTotalCount.asInstanceOf[Number].longValue(), if(avgAreaDotTotalCount==null) 0L else avgAreaDotTotalCount.asInstanceOf[Number].longValue(), if(companyDotTotalCount.isNullAt(0)) 0L else ttDotTotalCount.get(0).asInstanceOf[Number].longValue(), if(maxCompanyDotTotalCount==null) 0L else maxCompanyDotTotalCount.asInstanceOf[Number].longValue(), if(minCompanyDotTotalCount==null) 0L else minCompanyDotTotalCount.asInstanceOf[Number].longValue(), if(avgCompanyDotTotalCount==null) 0L else avgCompanyDotTotalCount.asInstanceOf[Number].longValue() ) dotRows.append(rowInfo) println(rowInfo) ttDotDetailByDayDF.unpersist() ttDotTotalCountDF.unpersist() areaDotTotalCountDF.unpersist() companyDotTotalCountDF.unpersist() }) //根据仓库车辆的日期进行分组 val ttWsDetailGroupByDayDF: DataFrame = ttWarehouseDetailDF.select("day").groupBy("day").count().cache() //定义计算好的指标结果集合对象 val wsRows: ArrayBuffer[Row] = ArrayBuffer[Row]() //循环遍历仓库车辆每个日期的车辆明细宽表数据 ttWsDetailGroupByDayDF.collect().foreach(row=> { //获取到要处理的数据所在的日期 val day: String = row.getAs[String](0) //返回指定日期的仓库明细数据 val ttWsDetailByDayDF: DataFrame = ttWarehouseDetailDF.where(col("day") === day).toDF().persist(StorageLevel.DISK_ONLY_2) val whTransportToolTotalCountDF: DataFrame = ttWsDetailByDayDF.groupBy($"ws_id").agg(count("ws_id").alias("cnt")).persist(StorageLevel.DISK_ONLY_2) //计算各仓库的总发车次数 val whTransportToolTotalCount: Row = whTransportToolTotalCountDF.agg(sum("cnt")).first() //各仓库的最大发车次数、最小发车次数、平均发车次数 val maxAndMinAndAvgWhTransportToolTotalCount: Row = whTransportToolTotalCountDF.agg(max("cnt"), min("cnt"), avg("cnt")).first() val maxTransportToolCount: Any = maxAndMinAndAvgWhTransportToolTotalCount(0) val minTransportToolCount: Any = maxAndMinAndAvgWhTransportToolTotalCount(1) val avgTransportToolCount: Any = maxAndMinAndAvgWhTransportToolTotalCount(2) // 各区域发车次数 val areaTransportToolTotalCountDF = ttWsDetailByDayDF.groupBy("city_id").agg(count($"id").alias("cnt")).persist(StorageLevel.DISK_ONLY_2) val areaTransportToolTotalCount = areaTransportToolTotalCountDF.agg(sum($"cnt")).first() // 各区域最大发车次数、最小发车次数和平均发车次数 val maxAndMinAndAvgAreaTransportToolTotalCount = areaTransportToolTotalCountDF.agg(max($"cnt"), min($"cnt"), avg($"cnt")).first() val maxAreaTransportToolTotalCount: Any = maxAndMinAndAvgAreaTransportToolTotalCount(0) val minAreaTransportToolTotalCount = maxAndMinAndAvgAreaTransportToolTotalCount(1) val avgAreaTransportToolTotalCount = maxAndMinAndAvgAreaTransportToolTotalCount(2) // 各公司发车次数 val companyTransportToolTotalCountDF = ttWsDetailByDayDF.groupBy("company_id").agg(count($"id").alias("cnt")).persist(StorageLevel.DISK_ONLY_2) val companyTransportToolTotalCount = companyTransportToolTotalCountDF.agg(sum($"cnt")).first() // 各公司最大发车次数、最小发车次数和平均发车次数 val maxAndMinAndAvgCompanyTransportToolTotalCount = companyTransportToolTotalCountDF.agg(max($"cnt"), min($"cnt"), avg($"cnt")).first() val maxCompanyTransportToolTotalCount = maxAndMinAndAvgCompanyTransportToolTotalCount(0) val minCompanyTransportToolTotalCount = maxAndMinAndAvgCompanyTransportToolTotalCount(1) val avgCompanyTransportToolTotalCount = maxAndMinAndAvgCompanyTransportToolTotalCount(2) //将计算好的指标数据封装到row对象中 val rowInfo: Row = Row( day, if(whTransportToolTotalCount.isNullAt(0)) 0L else whTransportToolTotalCount.get(0).asInstanceOf[Number].longValue(), if(maxTransportToolCount==null) 0L else maxTransportToolCount.asInstanceOf[Number].longValue(), if(minTransportToolCount==null) 0L else minTransportToolCount.asInstanceOf[Number].longValue(), if(avgTransportToolCount==null) 0L else avgTransportToolCount.asInstanceOf[Number].longValue(), if(areaTransportToolTotalCount.isNullAt(0)) 0L else areaTransportToolTotalCount.get(0).asInstanceOf[Number].longValue(), if(maxAreaTransportToolTotalCount==null) 0L else maxAreaTransportToolTotalCount.asInstanceOf[Number].longValue(), if(minAreaTransportToolTotalCount==null) 0L else minAreaTransportToolTotalCount.asInstanceOf[Number].longValue(), if(avgAreaTransportToolTotalCount==null) 0L else avgAreaTransportToolTotalCount.asInstanceOf[Number].longValue(), if(companyTransportToolTotalCount.isNullAt(0)) 0L else companyTransportToolTotalCount.get(0).asInstanceOf[Number].longValue(), if(maxCompanyTransportToolTotalCount==null) 0L else maxCompanyTransportToolTotalCount.asInstanceOf[Number].longValue(), if(minCompanyTransportToolTotalCount==null) 0L else minCompanyTransportToolTotalCount.asInstanceOf[Number].longValue(), if(avgCompanyTransportToolTotalCount==null) 0L else avgCompanyTransportToolTotalCount.asInstanceOf[Number].longValue() ) wsRows.append(rowInfo) println(rowInfo) ttWsDetailByDayDF.unpersist() areaTransportToolTotalCountDF.unpersist() companyTransportToolTotalCountDF.unpersist() whTransportToolTotalCountDF.unpersist() })2.4、通过StructType构建指定Schema//定义指标结果表的shema信息 //网点车辆相关的表结构数据 val schemaDot: StructType = StructType(Array( StructField("id", StringType, false, Metadata.empty), StructField("ttDotTotalCount", LongType, true, Metadata.empty), //各网点发车次数 StructField("maxTtDotTotalCount", LongType, true, Metadata.empty), //各网点最大发车次数 StructField("minTtDotTotalCount", LongType, true, Metadata.empty), //各网点最小发车次数 StructField("avgTtDotTotalCount", LongType, true, Metadata.empty), //各网点平均发车次数 StructField("areaDotTotalCount", LongType, true, Metadata.empty), //各区域发车次数 StructField("maxAreaDotTotalCount", LongType, true, Metadata.empty), //各区域最大发车次数 StructField("minAreaDotTotalCount", LongType, true, Metadata.empty), //各区域最小发车次数 StructField("avgAreaDotTotalCount", LongType, true, Metadata.empty), //各区域平均发车次数 StructField("companyDotTotalCount", LongType, true, Metadata.empty), //各公司发车次数 StructField("maxCompanyDotTotalCount", LongType, true, Metadata.empty), //各公司最大发车次数 StructField("minCompanyDotTotalCount", LongType, true, Metadata.empty), //各公司最小发车次数 StructField("avgCompanyDotTotalCount", LongType, true, Metadata.empty) //各公司平均发车次数 )) //仓库车辆相关的表结构数据 val schemaWs: StructType = StructType(Array( StructField("id", StringType, false, Metadata.empty), StructField("whTransportToolTotalCount", LongType, true, Metadata.empty), //各仓库发车次数 StructField("maxWhTransportToolTotalCount", LongType, true, Metadata.empty), //各仓库最大发车次数 StructField("minWhTransportToolTotalCount", LongType, true, Metadata.empty), //各仓库最小发车次数 StructField("avgWhTransportToolTotalCount", LongType, true, Metadata.empty), //各仓库平均发车次数 StructField("areaTransportToolTotalCount", LongType, true, Metadata.empty), //各区域发车次数 StructField("maxAreaTransportToolTotalCount", LongType, true, Metadata.empty), //各区域最大发车次数 StructField("minAreaTransportToolTotalCount", LongType, true, Metadata.empty), //各区域最小发车次数 StructField("avgAreaTransportToolTotalCount", LongType, true, Metadata.empty), //各区域平均发车次数 StructField("companyTransportToolTotalCount", LongType, true, Metadata.empty), //各公司发车次数 StructField("maxCompanyTransportToolTotalCount", LongType, true, Metadata.empty), //各公司最大发车次数 StructField("minCompanyTransportToolTotalCount", LongType, true, Metadata.empty), //各公司最小发车次数 StructField("avgCompanyTransportToolTotalCount", LongType, true, Metadata.empty) //各公司平均发车次数 ))2.5、持久化指标数据到kudu表//TODO 5）将计算好的指标数据写入到kudu数据库中 val dotRDD: RDD[Row] = sparkSession.sparkContext.makeRDD(dotRows) val wsRDD: RDD[Row] = sparkSession.sparkContext.makeRDD(wsRows) val dotDataFrame: DataFrame = sparkSession.createDataFrame(dotRDD, schemaDot) val wsDataFrame: DataFrame = sparkSession.createDataFrame(wsRDD, schemaWs) save(dotDataFrame, OfflineTableDefine.ttDotSummary) save(wsDataFrame, OfflineTableDefine.ttWsSummary)2.6、完整代码package cn.it.logistics.offline.dws import cn.it.logistics.common.{Configuration, DateHelper, OfflineTableDefine, SparkUtils} import cn.it.logistics.offline.OfflineApp import org.apache.spark.SparkConf import org.apache.spark.rdd.RDD import org.apache.spark.sql.{DataFrame, Row, SparkSession} import org.apache.spark.sql.types.{DoubleType, LongType, Metadata, StringType, StructField, StructType} import org.apache.spark.sql.functions._ import org.apache.spark.storage.StorageLevel import scala.collection.mutable.ArrayBuffer /** * 车辆主题指标开发 */ object TransportToolDWS extends OfflineApp{ //定义应用程序的名称 val appName = this.getClass.getSimpleName /** * 入口函数 * @param args */ def main(args: Array[String]): Unit = { /** * 实现步骤： * 1）创建SparkConf对象 * 2）创建SparkSession对象 * 3）读取车辆明细宽表的数据 * 4）对车辆明细宽表的数据进行指标的计算 * 5）将计算好的指标数据写入到kudu数据库中 * 5.1：定义指标结果表的schema信息 * 5.2：组织需要写入到kudu表的数据 * 5.3：判断指标结果表是否存在，如果不存在则创建 * 5.4：将数据写入到kudu表中 * 6）删除缓存数据 * 7）停止任务，退出sparksession */ //TODO 1）创建SparkConf对象 val sparkConf: SparkConf = SparkUtils.autoSettingEnv( SparkUtils.sparkConf(appName) ) //TODO 2）创建SparkSession对象 val sparkSession: SparkSession = SparkUtils.getSparkSession(sparkConf) sparkSession.sparkContext.setLogLevel(Configuration.LOG_OFF) //处理数据 execute(sparkSession) } /** * 数据处理 * * @param sparkSession */ override def execute(sparkSession: SparkSession): Unit = { //TODO 3）读取车辆明细宽表的数据 val ttDotDetailDF: DataFrame = getKuduSource(sparkSession, OfflineTableDefine.dotTransportToolDetail, Configuration.isFirstRunnable).persist(StorageLevel.DISK_ONLY_2) val ttWarehouseDetailDF: DataFrame = getKuduSource(sparkSession, OfflineTableDefine.warehouseTransportToolDetail, Configuration.isFirstRunnable).persist(StorageLevel.DISK_ONLY_2) //根据网点车辆的日期进行分组 val ttDotDetailGroupByDayDF: DataFrame = ttDotDetailDF.select("day").groupBy("day").count().cache() //导入隐式转换 import sparkSession.implicits._ //定义计算好的指标结果集合对象 val dotRows: ArrayBuffer[Row] = ArrayBuffer[Row]() //循环遍历网点车辆每个日期的车辆明细宽表数据 ttDotDetailGroupByDayDF.collect().foreach(row=> { //获取到要处理的数据所在的日期 val day: String = row.getAs[String](0) //返回指定日期的仓库明细数据 val ttDotDetailByDayDF: DataFrame = ttDotDetailDF.where(col("day") === day).toDF().persist(StorageLevel.DISK_ONLY_2) //各网点的发车次数（西三旗：10，西二旗：20） val ttDotTotalCountDF: DataFrame = ttDotDetailByDayDF.groupBy($"dot_id").agg(count("dot_id").alias("cnt")).persist(StorageLevel.DISK_ONLY_2) //计算各网点的总发车次数 val ttDotTotalCount: Row = ttDotTotalCountDF.agg(sum("cnt")).first() //各网点的最大发车次数、最小发车次数、平均发车次数 val maxAndMinAndAvgTTDotTotalCount: Row = ttDotTotalCountDF.agg(max("cnt"), min("cnt"), avg("cnt")).first() val maxTTDotTotalCount: Any = maxAndMinAndAvgTTDotTotalCount(0) val minTTDotTotalCount: Any = maxAndMinAndAvgTTDotTotalCount(1) val avgTTDotTotalCount: Any = maxAndMinAndAvgTTDotTotalCount(2) // 各区域发车次数 val areaDotTotalCountDF = ttDotDetailByDayDF.groupBy("city_id").agg(count($"id").alias("cnt")).persist(StorageLevel.DISK_ONLY_2) val areaDotTotalCount = areaDotTotalCountDF.agg(sum($"cnt")).first() // 各区域最大发车次数、最小发车次数和平均发车次数 val maxAndMinAndAvgAreaDotTotalCount = areaDotTotalCountDF.agg(max($"cnt"), min($"cnt"), avg($"cnt")).first() val maxAreaDotTotalCount = maxAndMinAndAvgAreaDotTotalCount(0) val minAreaDotTotalCount = maxAndMinAndAvgAreaDotTotalCount(1) val avgAreaDotTotalCount = maxAndMinAndAvgAreaDotTotalCount(2) // 各公司发车次数 val companyDotTotalCountDF = ttDotDetailByDayDF.groupBy("company_id").agg(count($"id").alias("cnt")).persist(StorageLevel.DISK_ONLY_2) val companyDotTotalCount = companyDotTotalCountDF.agg(sum($"cnt")).first() // 各公司最大发车次数、最小发车次数和平均发车次数 val maxAndMinAndAvgCompanyDotTotalCount = companyDotTotalCountDF.agg(max($"cnt"), min($"cnt"), avg($"cnt")).first() val maxCompanyDotTotalCount = maxAndMinAndAvgCompanyDotTotalCount(0) val minCompanyDotTotalCount = maxAndMinAndAvgCompanyDotTotalCount(1) val avgCompanyDotTotalCount = maxAndMinAndAvgCompanyDotTotalCount(2) //将计算好的指标数据封装到row对象中 val rowInfo: Row = Row( day, if(ttDotTotalCount.isNullAt(0)) 0L else ttDotTotalCount.get(0).asInstanceOf[Number].longValue(), if(maxTTDotTotalCount==null) 0L else maxTTDotTotalCount.asInstanceOf[Number].longValue(), if(minTTDotTotalCount==null) 0L else minTTDotTotalCount.asInstanceOf[Number].longValue(), if(avgTTDotTotalCount==null) 0L else avgTTDotTotalCount.asInstanceOf[Number].longValue(), if(areaDotTotalCount.isNullAt(0)) 0L else areaDotTotalCount.get(0).asInstanceOf[Number].longValue(), if(maxAreaDotTotalCount==null) 0L else maxAreaDotTotalCount.asInstanceOf[Number].longValue(), if(minAreaDotTotalCount==null) 0L else minAreaDotTotalCount.asInstanceOf[Number].longValue(), if(avgAreaDotTotalCount==null) 0L else avgAreaDotTotalCount.asInstanceOf[Number].longValue(), if(companyDotTotalCount.isNullAt(0)) 0L else ttDotTotalCount.get(0).asInstanceOf[Number].longValue(), if(maxCompanyDotTotalCount==null) 0L else maxCompanyDotTotalCount.asInstanceOf[Number].longValue(), if(minCompanyDotTotalCount==null) 0L else minCompanyDotTotalCount.asInstanceOf[Number].longValue(), if(avgCompanyDotTotalCount==null) 0L else avgCompanyDotTotalCount.asInstanceOf[Number].longValue() ) dotRows.append(rowInfo) println(rowInfo) ttDotDetailByDayDF.unpersist() ttDotTotalCountDF.unpersist() areaDotTotalCountDF.unpersist() companyDotTotalCountDF.unpersist() }) //根据仓库车辆的日期进行分组 val ttWsDetailGroupByDayDF: DataFrame = ttWarehouseDetailDF.select("day").groupBy("day").count().cache() //定义计算好的指标结果集合对象 val wsRows: ArrayBuffer[Row] = ArrayBuffer[Row]() //循环遍历仓库车辆每个日期的车辆明细宽表数据 ttWsDetailGroupByDayDF.collect().foreach(row=> { //获取到要处理的数据所在的日期 val day: String = row.getAs[String](0) //返回指定日期的仓库明细数据 val ttWsDetailByDayDF: DataFrame = ttWarehouseDetailDF.where(col("day") === day).toDF().persist(StorageLevel.DISK_ONLY_2) val whTransportToolTotalCountDF: DataFrame = ttWsDetailByDayDF.groupBy($"ws_id").agg(count("ws_id").alias("cnt")).persist(StorageLevel.DISK_ONLY_2) //计算各仓库的总发车次数 val whTransportToolTotalCount: Row = whTransportToolTotalCountDF.agg(sum("cnt")).first() //各仓库的最大发车次数、最小发车次数、平均发车次数 val maxAndMinAndAvgWhTransportToolTotalCount: Row = whTransportToolTotalCountDF.agg(max("cnt"), min("cnt"), avg("cnt")).first() val maxTransportToolCount: Any = maxAndMinAndAvgWhTransportToolTotalCount(0) val minTransportToolCount: Any = maxAndMinAndAvgWhTransportToolTotalCount(1) val avgTransportToolCount: Any = maxAndMinAndAvgWhTransportToolTotalCount(2) // 各区域发车次数 val areaTransportToolTotalCountDF = ttWsDetailByDayDF.groupBy("city_id").agg(count($"id").alias("cnt")).persist(StorageLevel.DISK_ONLY_2) val areaTransportToolTotalCount = areaTransportToolTotalCountDF.agg(sum($"cnt")).first() // 各区域最大发车次数、最小发车次数和平均发车次数 val maxAndMinAndAvgAreaTransportToolTotalCount = areaTransportToolTotalCountDF.agg(max($"cnt"), min($"cnt"), avg($"cnt")).first() val maxAreaTransportToolTotalCount: Any = maxAndMinAndAvgAreaTransportToolTotalCount(0) val minAreaTransportToolTotalCount = maxAndMinAndAvgAreaTransportToolTotalCount(1) val avgAreaTransportToolTotalCount = maxAndMinAndAvgAreaTransportToolTotalCount(2) // 各公司发车次数 val companyTransportToolTotalCountDF = ttWsDetailByDayDF.groupBy("company_id").agg(count($"id").alias("cnt")).persist(StorageLevel.DISK_ONLY_2) val companyTransportToolTotalCount = companyTransportToolTotalCountDF.agg(sum($"cnt")).first() // 各公司最大发车次数、最小发车次数和平均发车次数 val maxAndMinAndAvgCompanyTransportToolTotalCount = companyTransportToolTotalCountDF.agg(max($"cnt"), min($"cnt"), avg($"cnt")).first() val maxCompanyTransportToolTotalCount = maxAndMinAndAvgCompanyTransportToolTotalCount(0) val minCompanyTransportToolTotalCount = maxAndMinAndAvgCompanyTransportToolTotalCount(1) val avgCompanyTransportToolTotalCount = maxAndMinAndAvgCompanyTransportToolTotalCount(2) //将计算好的指标数据封装到row对象中 val rowInfo: Row = Row( day, if(whTransportToolTotalCount.isNullAt(0)) 0L else whTransportToolTotalCount.get(0).asInstanceOf[Number].longValue(), if(maxTransportToolCount==null) 0L else maxTransportToolCount.asInstanceOf[Number].longValue(), if(minTransportToolCount==null) 0L else minTransportToolCount.asInstanceOf[Number].longValue(), if(avgTransportToolCount==null) 0L else avgTransportToolCount.asInstanceOf[Number].longValue(), if(areaTransportToolTotalCount.isNullAt(0)) 0L else areaTransportToolTotalCount.get(0).asInstanceOf[Number].longValue(), if(maxAreaTransportToolTotalCount==null) 0L else maxAreaTransportToolTotalCount.asInstanceOf[Number].longValue(), if(minAreaTransportToolTotalCount==null) 0L else minAreaTransportToolTotalCount.asInstanceOf[Number].longValue(), if(avgAreaTransportToolTotalCount==null) 0L else avgAreaTransportToolTotalCount.asInstanceOf[Number].longValue(), if(companyTransportToolTotalCount.isNullAt(0)) 0L else companyTransportToolTotalCount.get(0).asInstanceOf[Number].longValue(), if(maxCompanyTransportToolTotalCount==null) 0L else maxCompanyTransportToolTotalCount.asInstanceOf[Number].longValue(), if(minCompanyTransportToolTotalCount==null) 0L else minCompanyTransportToolTotalCount.asInstanceOf[Number].longValue(), if(avgCompanyTransportToolTotalCount==null) 0L else avgCompanyTransportToolTotalCount.asInstanceOf[Number].longValue() ) wsRows.append(rowInfo) println(rowInfo) ttWsDetailByDayDF.unpersist() areaTransportToolTotalCountDF.unpersist() companyTransportToolTotalCountDF.unpersist() whTransportToolTotalCountDF.unpersist() }) //定义指标结果表的shema信息 //网点车辆相关的表结构数据 val schemaDot: StructType = StructType(Array( StructField("id", StringType, false, Metadata.empty), StructField("ttDotTotalCount", LongType, true, Metadata.empty), //各网点发车次数 StructField("maxTtDotTotalCount", LongType, true, Metadata.empty), //各网点最大发车次数 StructField("minTtDotTotalCount", LongType, true, Metadata.empty), //各网点最小发车次数 StructField("avgTtDotTotalCount", LongType, true, Metadata.empty), //各网点平均发车次数 StructField("areaDotTotalCount", LongType, true, Metadata.empty), //各区域发车次数 StructField("maxAreaDotTotalCount", LongType, true, Metadata.empty), //各区域最大发车次数 StructField("minAreaDotTotalCount", LongType, true, Metadata.empty), //各区域最小发车次数 StructField("avgAreaDotTotalCount", LongType, true, Metadata.empty), //各区域平均发车次数 StructField("companyDotTotalCount", LongType, true, Metadata.empty), //各公司发车次数 StructField("maxCompanyDotTotalCount", LongType, true, Metadata.empty), //各公司最大发车次数 StructField("minCompanyDotTotalCount", LongType, true, Metadata.empty), //各公司最小发车次数 StructField("avgCompanyDotTotalCount", LongType, true, Metadata.empty) //各公司平均发车次数 )) //仓库车辆相关的表结构数据 val schemaWs: StructType = StructType(Array( StructField("id", StringType, false, Metadata.empty), StructField("whTransportToolTotalCount", LongType, true, Metadata.empty), //各仓库发车次数 StructField("maxWhTransportToolTotalCount", LongType, true, Metadata.empty), //各仓库最大发车次数 StructField("minWhTransportToolTotalCount", LongType, true, Metadata.empty), //各仓库最小发车次数 StructField("avgWhTransportToolTotalCount", LongType, true, Metadata.empty), //各仓库平均发车次数 StructField("areaTransportToolTotalCount", LongType, true, Metadata.empty), //各区域发车次数 StructField("maxAreaTransportToolTotalCount", LongType, true, Metadata.empty), //各区域最大发车次数 StructField("minAreaTransportToolTotalCount", LongType, true, Metadata.empty), //各区域最小发车次数 StructField("avgAreaTransportToolTotalCount", LongType, true, Metadata.empty), //各区域平均发车次数 StructField("companyTransportToolTotalCount", LongType, true, Metadata.empty), //各公司发车次数 StructField("maxCompanyTransportToolTotalCount", LongType, true, Metadata.empty), //各公司最大发车次数 StructField("minCompanyTransportToolTotalCount", LongType, true, Metadata.empty), //各公司最小发车次数 StructField("avgCompanyTransportToolTotalCount", LongType, true, Metadata.empty) //各公司平均发车次数 )) //TODO 5）将计算好的指标数据写入到kudu数据库中 val dotRDD: RDD[Row] = sparkSession.sparkContext.makeRDD(dotRows) val wsRDD: RDD[Row] = sparkSession.sparkContext.makeRDD(wsRows) val dotDataFrame: DataFrame = sparkSession.createDataFrame(dotRDD, schemaDot) val wsDataFrame: DataFrame = sparkSession.createDataFrame(wsRDD, schemaWs) save(dotDataFrame, OfflineTableDefine.ttDotSummary) save(wsDataFrame, OfflineTableDefine.ttWsSummary) // 6）删除缓存数据 ttDotDetailDF.unpersist() ttWarehouseDetailDF.unpersist() ttDotDetailGroupByDayDF.unpersist() ttWsDetailGroupByDayDF.unpersist() // 7）停止任务，退出sparksession sparkSession.stop() } }

文章

缓存 · 分布式计算 · 大数据 · 调度 · Spark · SQL

2022-05-19

以Java的视角来聊聊BIO、NIO与AIO的区别？

1F说一说I/O首先来说一下什么是I/O?在计算机系统中I/O就是输入（Input）和输出(Output)的意思，针对不同的操作对象，可以划分为磁盘I/O模型，网络I/O模型，内存映射I/O, Direct I/O、数据库I/O等，只要具有输入输出类型的交互系统都可以认为是I/O系统，也可以说I/O是整个操作系统数据交换与人机交互的通道，这个概念与选用的开发语言没有关系，是一个通用的概念。在如今的系统中I/O却拥有很重要的位置，现在系统都有可能处理大量文件，大量数据库操作，而这些操作都依赖于系统的I/O性能，也就造成了现在系统的瓶颈往往都是由于I/O性能造成的。因此，为了解决磁盘I/O性能慢的问题，系统架构中添加了缓存来提高响应速度；或者有些高端服务器从硬件级入手，使用了固态硬盘（SSD）来替换传统机械硬盘；在大数据方面，Spark越来越多的承担了实时性计算任务，而传统的Hadoop体系则大多应用在了离线计算与大量数据存储的场景，这也是由于磁盘I/O性能远不如内存I/O性能而造成的格局（Spark更多的使用了内存，而MapReduece更多的使用了磁盘）。因此，一个系统的优化空间，往往都在低效率的I/O环节上，很少看到一个系统CPU、内存的性能是其整个系统的瓶颈。也正因为如此，Java在I/O上也一直在做持续的优化，从JDK 1.4开始便引入了NIO模型，大大的提高了以往BIO模型下的操作效率。这里先给出BIO、NIO、AIO的基本定义与类比描述：BIO （Blocking I/O）：同步阻塞I/O模式，数据的读取写入必须阻塞在一个线程内等待其完成。这里使用那个经典的烧开水例子，这里假设一个烧开水的场景，有一排水壶在烧开水，BIO的工作模式就是， 叫一个线程停留在一个水壶那，直到这个水壶烧开，才去处理下一个水壶。但是实际上线程在等待水壶烧开的时间段什么都没有做。NIO （New I/O）：同时支持阻塞与非阻塞模式，但这里我们以其同步非阻塞I/O模式来说明，那么什么叫做同步非阻塞？如果还拿烧开水来说，NIO的做法是叫一个线程不断的轮询每个水壶的状态，看看是否有水壶的状态发生了改变，从而进行下一步的操作。AIO （ Asynchronous I/O）：异步非阻塞I/O模型。异步非阻塞与同步非阻塞的区别在哪里？异步非阻塞无需一个线程去轮询所有IO操作的状态改变，在相应的状态改变后，系统会通知对应的线程来处理。对应到烧开水中就是，为每个水壶上面装了一个开关，水烧开之后，水壶会自动通知我水烧开了。进程中的IO调用步骤大致可以分为以下四步：?进程向操作系统请求数据 ;操作系统把外部数据加载到内核的缓冲区中; 操作系统把内核的缓冲区拷贝到进程的缓冲区 ;进程获得数据完成自己的功能 ;当操作系统在把外部数据放到进程缓冲区的这段时间（即上述的第二，三步），如果应用进程是挂起等待的，那么就是同步IO，反之，就是异步IO，也就是AIO 。2FBIO（Blocking I/O）同步阻塞I/O这是最基本与简单的I/O操作方式，其根本特性是做完一件事再去做另一件事，一件事一定要等前一件事做完，这很符合程序员传统的顺序来开发思想，因此BIO模型程序开发起来较为简单，易于把握。但是BIO如果需要同时做很多事情（例如同时读很多文件，处理很多tcp请求等），就需要系统创建很多线程来完成对应的工作，因为BIO模型下一个线程同时只能做一个工作，如果线程在执行过程中依赖于需要等待的资源，那么该线程会长期处于阻塞状态，我们知道在整个操作系统中，线程是系统执行的基本单位，在BIO模型下的线程阻塞就会导致系统线程的切换，从而对整个系统性能造成一定的影响。当然如果我们只需要创建少量可控的线程，那么采用BIO模型也是很好的选择，但如果在需要考虑高并发的web或者tcp服务器中采用BIO模型就无法应对了，如果系统开辟成千上万的线程，那么CPU的执行时机都会浪费在线程的切换中，使得线程的执行效率大大降低。此外，关于线程这里说一句题外话，在系统开发中线程的生命周期一定要准确控制，在需要一定规模并发的情形下，尽量使用线程池来确保线程创建数目在一个合理的范围之内，切莫编写线程数量创建上限的代码。3FNIO (New I/O) 同步非阻塞I/O关于NIO，国内有很多技术博客将英文翻译成No-Blocking I/O，非阻塞I/O模型 ，当然这样就与BIO形成了鲜明的特性对比。NIO本身是基于事件驱动的思想来实现的，其目的就是解决BIO的大并发问题，在BIO模型中，如果需要并发处理多个I/O请求，那就需要多线程来支持，NIO使用了多路复用器机制，以socket使用来说，多路复用器通过不断轮询各个连接的状态，只有在socket有流可读或者可写时，应用程序才需要去处理它，在线程的使用上，就不需要一个连接就必须使用一个处理线程了，而是只是有效请求时（确实需要进行I/O处理时），才会使用一个线程去处理，这样就避免了BIO模型下大量线程处于阻塞等待状态的情景。相对于BIO的流，NIO抽象出了新的通道（Channel）作为输入输出的通道，并且提供了缓存（Buffer）的支持，在进行读操作时，需要使用Buffer分配空间，然后将数据从Channel中读入Buffer中，对于Channel的写操作，也需要现将数据写入Buffer，然后将Buffer写入Channel中。如下是NIO方式进行文件拷贝操作的示例，见下图：通过比较New IO的使用方式我们可以发现，新的IO操作不再面向 Stream来进行操作了，改为了通道Channel，并且使用了更加灵活的缓存区类Buffer，Buffer只是缓存区定义接口， 根据需要，我们可以选择对应类型的缓存区实现类。在java NIO编程中，我们需要理解以下3个对象Channel、Buffer和Selector。Channel首先说一下Channel，国内大多翻译成“通道”。Channel和IO中的Stream(流)是差不多一个等级的。只不过Stream是单向的，譬如：InputStream, OutputStream。而Channel是双向的，既可以用来进行读操作，又可以用来进行写操作，NIO中的Channel的主要实现有：FileChannel、DatagramChannel、SocketChannel、ServerSocketChannel；通过看名字就可以猜出个所以然来：分别可以对应文件IO、UDP和TCP（Server和Client）。BufferNIO中的关键Buffer实现有：ByteBuffer、CharBuffer、DoubleBuffer、 FloatBuffer、IntBuffer、 LongBuffer,、ShortBuffer，分别对应基本数据类型: byte、char、double、 float、int、 long、 short。当然NIO中还有MappedByteBuffer, HeapByteBuffer, DirectByteBuffer等这里先不具体陈述其用法细节。说一下 DirectByteBuffer 与 HeapByteBuffer 的区别？它们 ByteBuffer 分配内存的两种方式。HeapByteBuffer 顾名思义其内存空间在 JVM 的 heap（堆）上分配，可以看做是 jdk 对于 byte[] 数组的封装；而 DirectByteBuffer 则直接利用了系统接口进行内存申请，其内存分配在c heap 中，这样就减少了内存之间的拷贝操作，如此一来，在使用 DirectByteBuffer 时，系统就可以直接从内存将数据写入到 Channel 中，而无需进行 Java 堆的内存申请，复制等操作，提高了性能。既然如此，为什么不直接使用 DirectByteBuffer，还要来个 HeapByteBuffer？原因在于， DirectByteBuffer 是通过full gc来回收内存的，DirectByteBuffer会自己检测情况而调用 system.gc()，但是如果参数中使用了 DisableExplicitGC 那么就无法回收该快内存了，-XX:+DisableExplicitGC标志自动将 System.gc() 调用转换成一个空操作，就是应用中调用 System.gc() 会变成一个空操作，那么如果设置了就需要我们手动来回收内存了，所以DirectByteBuffer使用起来相对于完全托管于 java 内存管理的Heap ByteBuffer 来说更复杂一些，如果用不好可能会引起OOM。Direct ByteBuffer 的内存大小受 -XX:MaxDirectMemorySize JVM 参数控制（默认大小64M），在 DirectByteBuffer 申请内存空间达到该设置大小后，会触发 Full GC。SelectorSelector 是NIO相对于BIO实现多路复用的基础，Selector 运行单线程处理多个 Channel，如果你的应用打开了多个通道，但每个连接的流量都很低，使用 Selector 就会很方便。例如在一个聊天服务器中。要使用 Selector , 得向 Selector 注册 Channel，然后调用它的 select() 方法。这个方法会一直阻塞到某个注册的通道有事件就绪。一旦这个方法返回，线程就可以处理这些事件，事件的例子有如新的连接进来、数据接收等。这里我们再来看一个NIO模型下的TCP服务器的实现，我们可以看到Selector 正是NIO模型下 TCP Server 实现IO复用的关键，请仔细理解下段代码while循环中的逻辑，见下图：4FAIO (Asynchronous I/O) 异步非阻塞I/OJava AIO就是Java作为对异步IO提供支持的NIO.2 ，Java NIO2 (JSR 203)定义了更多的 New I/O APIs， 提案2003提出，直到2011年才发布， 最终在JDK 7中才实现。JSR 203除了提供更多的文件系统操作API(包括可插拔的自定义的文件系统)， 还提供了对socket和文件的异步 I/O操作。同时实现了JSR-51提案中的socket channel全部功能,包括对绑定， option配置的支持以及多播multicast的实现。从编程模式上来看AIO相对于NIO的区别在于，NIO需要使用者线程不停的轮询IO对象，来确定是否有数据准备好可以读了，而AIO则是在数据准备好之后，才会通知数据使用者，这样使用者就不需要不停地轮询了。当然AIO的异步特性并不是Java实现的伪异步，而是使用了系统底层API的支持，在Unix系统下，采用了epoll IO模型，而windows便是使用了IOCP模型。关于Java AIO，本篇只做一个抛砖引玉的介绍，如果你在实际工作中用到了，那么可以参考Netty在高并发下使用AIO的相关技术。总结? ??IO实质上与线程没有太多的关系，但是不同的IO模型改变了应用程序使用线程的方式，NIO与BIO的出现解决了很多BIO无法解决的并发问题，当然任何技术抛开适用场景都是耍流氓，复杂的技术往往是为了解决简单技术无法解决的问题而设计的，在系统开发中能用常规技术解决的问题，绝不用复杂技术，否则大大增加系统代码的维护难度，学习IT技术不是为了炫技，而是要实实在在解决问题。

文章

缓存 · 分布式计算 · 网络协议 · 固态存储 · Java · 程序员 · API · 数据库 · Spark · Windows

2022-05-21

Linux命令行终端切换工具-Screen

1简介Screen是一款由GNU计划开发的用于命令行终端切换的自由软件。用户可以通过该软件同时连接多个本地或远程的命令行会话，并在其间自由切换。GNU Screen可以看作是窗口管理器的命令行界面版本。它提供了统一的管理多个会话的界面和相应的功能。2会话恢复只要Screen本身没有终止，在其内部运行的会话都可以恢复。对于远程登录的用户特别有用——即使网络连接中断，用户也不会失去对已经打开的命令行会话的控制。只要再次登录到主机上执行screen -r就可以恢复会话的运行。同样在暂时离开的时候，也可以执行分离命令detach，在保证里面的程序正常运行的情况下让Screen挂起（切换到后台）。3多窗口在Screen环境下，所有的会话都独立的运行，并拥有各自的编号、输入、输出和窗口缓存。用户可以通过快捷键在不同的窗口下切换，并可以自由的重定向各个窗口的输入和输出。4会话共享Screen可以让一个或多个用户从不同终端多次登录一个会话，并共享会话的所有特性（比如可以看到完全相同的输出）。它同时提供了窗口访问权限的机制，可以对窗口进行密码保护。5怎么使用？命令说明：相关参数使用，我们可以通过screen -h 或者 man screen 查看，具体参数说明见下表：说了这么多命令参数，接下来给大家分享一些我经常用的参数，供参考，见下表:我建议小伙伴们还是要记住它的常用快捷键，具体见下表：6总结Screen 给我们再运行脚本和远程访问终端的时候，大大方便了我们的工作，更多牛逼的使用方法，需要慢慢摸索。

文章

缓存 · Linux

2022-05-21