]]>
本文介绍 Spark 的 CI 与 CD & CD 参考方法。包含如何维护源代码，如何维护 Release 多版本，开发版与正式版，以及如何实现灰度发布，如何进行 hotfix。
技术世界http://www.jasongj.com/spark/committer/2018-09-25T23:20:13.000Z2018-09-25T23:20:13.000Z
CREATETABLE score ( idINT, math_score INT, english_score INT )
1 2 3 4 5
CREATETABLE people ( idINT, age INT, nameINT )
对其进行关联查询如下
1 2 3 4 5 6 7 8 9
SELECTsum(v) FROM ( SELECT score.id, 100 + 80 + score.math_score + score.english_score AS v FROM people JOIN score ON people.id = score.id AND people.age > 10 ) tmp
This class provides thread-local variables. These variables differ from their normal counterparts in that each thread that accesses one (via its get or set method) has its own, independently initialized copy of the variable. ThreadLocal instances are typically private static fields in classes that wish to associate state with a thread (e.g., a user ID or Transaction ID). Each thread holds an implicit reference to its copy of a thread-local variable as long as the thread is alive and the ThreadLocal instance is accessible; after a thread goes away, all of its copies of thread-local instances are subject to garbage collection (unless other references to these copies exist).
public T get(){ Thread t = Thread.currentThread(); ThreadLocalMap map = getMap(t); if (map != null) { ThreadLocalMap.Entry e = map.getEntry(this); if (e != null) { @SuppressWarnings("unchecked") T result = (T)e.value; return result; } } return setInitialValue(); }
构造KTable时，需要指定其state store name。默认情况下，该名字也即用于存储该KTable的状态的Topic的名字，遍历KTable的过程，实际就是遍历它对应的state store，或者说遍历Topic的所有key，并取每个Key最新值的过程。为了使得该过程更加高效，默认情况下会对该Topic进行compact操作。
另外，除了KTable，所有状态计算，都需要指定state store name，从而记录中间状态。
基于ISR的数据复制方案 如《 Kafka High Availability（上）》一文所述，Kafka的数据复制是以Partition为单位的。而多个备份间的数据复制，通过Follower向Leader拉取数据完成。从一这点来讲，Kafka的数据复制方案接近于上文所讲的Master-Slave方案。不同的是，Kafka既不是完全的同步复制，也不是完全的异步复制，而是基于ISR的动态复制方案。
/** * Delete this log segment from the filesystem. * * @throws KafkaStorageException if the delete fails. */ defdelete() { val deletedLog = log.delete() val deletedIndex = index.delete() val deletedTimeIndex = timeIndex.delete() if(!deletedLog && log.file.exists) thrownewKafkaStorageException("Delete of log " + log.file.getName + " failed.") if(!deletedIndex && index.file.exists) thrownewKafkaStorageException("Delete of index " + index.file.getName + " failed.") if(!deletedTimeIndex && timeIndex.file.exists) thrownewKafkaStorageException("Delete of time index " + timeIndex.file.getName + " failed.") }
Linux 2.4+内核通过sendfile系统调用，提供了零拷贝。数据通过DMA拷贝到内核态Buffer后，直接通过DMA拷贝到NIC Buffer，无需CPU拷贝。这也是零拷贝这一说法的来源。除了减少数据拷贝外，因为整个读文件-网络发送由一个sendfile调用完成，整个过程只有两次上下文切换，因此大大提高了性能。零拷贝过程如下图所示。
INSERT OVERWRITE TABLE test SELECT CAST(CASE WHEN id < 908000000 THEN (9500000 + (CAST (RAND() * 2 AS INT) + 1) * 48 ) ELSE CAST(id/100 AS INT) END AS STRING), name FROM student_external WHERE id BETWEEN 900000000 AND 1050000000;
INSERT OVERWRITE TABLE test_new SELECT CAST(CAST(id/100 AS INT) AS STRING), name FROM student_delta_external WHERE id BETWEEN 950000000 AND 950500000;
int coreNum = Runtime.getRuntime().availableProcessors(); Processor[] processors = new Processor[coreNum]; for (int i = 0; i < processors.length; i++) { processors[i] = new Processor(); }
int index = 0; while (selector.select() > 0) { Set<SelectionKey> keys = selector.selectedKeys(); for (SelectionKey key : keys) { keys.remove(key); if (key.isAcceptable()) { ServerSocketChannel acceptServerSocketChannel = (ServerSocketChannel) key.channel(); SocketChannel socketChannel = acceptServerSocketChannel.accept(); socketChannel.configureBlocking(false); LOGGER.info("Accept request from {}", socketChannel.getRemoteAddress()); Processor processor = processors[(int) ((index++) % coreNum)]; processor.addChannel(socketChannel); processor.wakeup(); } } } } }
Statement st1 = xaConnection1.createStatement(); Statement st2 = xaConnection2.createStatement(); LOGGER.info("Connection autocommit after created statement: {}", xaConnection1.getAutoCommit());
st1.execute("update casp.test set qtime=current_timestamp, value = 1"); st2.execute("update casp.test set qtime=current_timestamp, value = 2"); LOGGER.info("Autocommit after execution : ", xaConnection1.getAutoCommit());
userTransaction.commit(); LOGGER.info("Autocommit after commit: ", xaConnection1.getAutoCommit()); return"commit";
publicclassCountDownLatchDemo{ publicstaticvoidmain(String[] args)throws InterruptedException { int totalThread = 3; long start = System.currentTimeMillis(); CountDownLatch countDown = new CountDownLatch(totalThread); for(int i = 0; i < totalThread; i++) { final String threadName = "Thread " + i; new Thread(() -> { System.out.println(String.format("%s\t%s %s", new Date(), threadName, "started")); try { Thread.sleep(1000); } catch (Exception ex) { ex.printStackTrace(); } countDown.countDown(); System.out.println(String.format("%s\t%s %s", new Date(), threadName, "ended")); }).start();; } countDown.await(); long stop = System.currentTimeMillis(); System.out.println(String.format("Total time : %sms", (stop - start))); } }
执行结果
1 2 3 4 5 6 7
Sun Jun 19 20:34:31 CST 2016 Thread 1 started Sun Jun 19 20:34:31 CST 2016 Thread 0 started Sun Jun 19 20:34:31 CST 2016 Thread 2 started Sun Jun 19 20:34:32 CST 2016 Thread 2 ended Sun Jun 19 20:34:32 CST 2016 Thread 1 ended Sun Jun 19 20:34:32 CST 2016 Thread 0 ended Total time : 1072ms
publicstaticvoidmain(String[] args){ int totalThread = 5; CyclicBarrier barrier = new CyclicBarrier(totalThread); for(int i = 0; i < totalThread; i++) { String threadName = "Thread " + i; new Thread(() -> { System.out.println(String.format("%s\t%s %s", new Date(), threadName, " is waiting")); try { barrier.await(); } catch (Exception ex) { ex.printStackTrace(); } System.out.println(String.format("%s\t%s %s", new Date(), threadName, "ended")); }).start(); } } }
执行结果如下
1 2 3 4 5 6 7 8 9 10
Sun Jun 19 21:04:49 CST 2016 Thread 1 is waiting Sun Jun 19 21:04:49 CST 2016 Thread 0 is waiting Sun Jun 19 21:04:49 CST 2016 Thread 3 is waiting Sun Jun 19 21:04:49 CST 2016 Thread 2 is waiting Sun Jun 19 21:04:49 CST 2016 Thread 4 is waiting Sun Jun 19 21:04:49 CST 2016 Thread 4 ended Sun Jun 19 21:04:49 CST 2016 Thread 0 ended Sun Jun 19 21:04:49 CST 2016 Thread 2 ended Sun Jun 19 21:04:49 CST 2016 Thread 1 ended Sun Jun 19 21:04:49 CST 2016 Thread 3 ended
tryLock(long time, TimeUnit unit) 该方法试图获得锁，若该锁当前可用，则立即获得锁并立即返回true。若锁当前不可用，则等待相应的时间（由该方法的两个参数决定）：1）若该时间内锁可用，则获得锁，并返回true；2）若等待期间当前线程被打断，则抛出InterruptedException；3）若等待时间结束仍未获得锁，则返回false。
重入锁可定义为公平锁或非公平锁，默认实现为非公平锁。
公平锁是指多个线程获取锁被阻塞的情况下，锁变为可用时，最新申请锁的线程获得锁。可通过在重入锁（RenentrantLock）的构造方法中传入true构建公平锁，如Lock lock = new RenentrantLock(true)
"thread-test-f" #18 prio=5 os_prio=31 tid=0x00007fefaa9b2800 nid=0x6603 waiting on condition [0x0000700002a3c000] java.lang.Thread.State: WAITING (parking) at sun.misc.Unsafe.park(Native Method) - parking to wait for <0x000000076af86810> (a java.util.concurrent.locks.ReentrantLock$FairSync) at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175) at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836) at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireQueued(AbstractQueuedSynchronizer.java:870) at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(AbstractQueuedSynchronizer.java:1199) at java.util.concurrent.locks.ReentrantLock$FairSync.lock(ReentrantLock.java:224) at java.util.concurrent.locks.ReentrantLock.lock(ReentrantLock.java:285) at com.jasongj.demo.TestJstack.lambda$5(TestJstack.java:69) at com.jasongj.demo.TestJstack$$Lambda$6/33524623.run(Unknown Source) at java.lang.Thread.run(Thread.java:745) Locked ownable synchronizers: - None
Sat Jun 18 21:33:46 CST 2016 Thread 1 started with read lock Sat Jun 18 21:33:46 CST 2016 Thread 2 started with read lock Sat Jun 18 21:33:48 CST 2016 Thread 2 ended Sat Jun 18 21:33:48 CST 2016 Thread 1 ended Sat Jun 18 21:33:48 CST 2016 Thread 3 started with write lock Sat Jun 18 21:33:50 CST 2016 Thread 3 ended
Sun Jun 19 15:59:09 CST 2016 Thread 1 is waiting Sun Jun 19 15:59:09 CST 2016 Thread 2 is running Sun Jun 19 15:59:13 CST 2016 Thread 2 ended Sun Jun 19 15:59:13 CST 2016 Thread 1 remaining time -2003467560 Sun Jun 19 15:59:13 CST 2016 Thread 1 is waken up
The results of a write by one thread are guaranteed to be visible to a read by another thread only if the write operation happens-before the read operation. The synchronized and volatile constructs, as well as the Thread.start() and Thread.join() methods, can form happens-before relationships.
Single Responsibility Principle（SRP），单一职责原则。它讲的是，不要存在多于一个导致类变更的原因，是高内聚低耦合的一个体现。
迪米特法则/最少知道原则
Law of Demeter or Least Knowledge Principle（LoD or LKP），迪米特法则或最少知道原则。它讲的是“一个对象就尽可能少的去了解其它对象”，从而实现松耦合。如果一个类的职责过多，由于多个职责耦合在了一起，任何一个职责的变更都可能引起其它职责的问题，严重影响了代码的可维护性和可重用性。
有些人认为，在某些特定场景下，卡塔尔世界杯BOB体育官方APP登入并非最优方案，而自己的解决方案可能会更好。这个问题得分两个方面来讨论：一方面，如上文所述，所有卡塔尔世界杯BOB体育官方APP登入都有其适用场景，“one size does not fit all”；另一方面，确实有可能自己设计的方案比卡塔尔世界杯BOB体育官方APP登入更适合，但这并不影响你学习并使用卡塔尔世界杯BOB体育官方APP登入，因为卡塔尔世界杯BOB体育官方APP登入经过大量实战检验能在绝大多数情况下提供良好方案。
privatestaticvoidtestJDKDynamicCreation(){ long start = System.currentTimeMillis(); for (int i = 0; i < creation; i++) { InvocationHandler handler = new SubjectProxyHandler(ConcreteSubject.class); Proxy.newProxyInstance(DynamicProxyPerfClient.class.getClassLoader(), new Class[] {ISubject.class}, handler); } long stop = System.currentTimeMillis(); LOG.info("JDK creation time : {} ms", stop - start); }
privatestaticvoidtestJDKDynamicExecution(){ long start = System.currentTimeMillis(); InvocationHandler handler = new SubjectProxyHandler(ConcreteSubject.class); ISubject subject = (ISubject) Proxy.newProxyInstance(DynamicProxyPerfClient.class.getClassLoader(), new Class[] {ISubject.class}, handler); for (int i = 0; i < execution; i++) { subject.action(); } long stop = System.currentTimeMillis(); LOG.info("JDK execution time : {} ms", stop - start); }
privatestaticvoidtestCglibCreation(){ long start = System.currentTimeMillis(); for (int i = 0; i < creation; i++) { MethodInterceptor methodInterceptor = new SubjectInterceptor(); Enhancer enhancer = new Enhancer(); enhancer.setSuperclass(ConcreteSubject.class); enhancer.setCallback(methodInterceptor); enhancer.create(); } long stop = System.currentTimeMillis(); LOG.info("cglib creation time : {} ms", stop - start); }
privatestaticvoidtestCglibExecution(){ MethodInterceptor methodInterceptor = new SubjectInterceptor(); Enhancer enhancer = new Enhancer(); enhancer.setSuperclass(ConcreteSubject.class); enhancer.setCallback(methodInterceptor); ISubject subject = (ISubject) enhancer.create(); long start = System.currentTimeMillis(); for (int i = 0; i < execution; i++) { subject.action(); } long stop = System.currentTimeMillis(); LOG.info("cglib execution time : {} ms", stop - start); }
}
结果如下
1 2 3 4
JDK creation time : 9924 ms JDK execution time : 3472 ms cglib creation time : 16108 ms cglib execution time : 6309 ms
publicstaticvoidmain(String[] args){ IDaoFactory factory = new MySQLDaoFactory();
IUserDao userDao = factory.createUserDao(); User user = new User(); user.setUsername("demo"); user.setPassword("demo".toCharArray()); userDao.addUser(user);
WITH regional_sales AS ( SELECT region, SUM(amount) AS total_sales FROM orders GROUPBY region ), top_regions AS ( SELECT region FROM regional_sales WHERE total_sales > (SELECTSUM(total_sales)/10FROM regional_sales ) SELECT region, product, SUM(quantity) AS product_units, SUM(amount) AS product_sales FROM orders WHERE region IN (SELECT region FROM top_regions) GROUPBY region, product;
WITH RECURSIVE result AS ( SELECCT id, name FROM chinamap WHERE id = 11 UNION ALL SELECT origin.id, result.name || ' > ' || origin.name FROMresult JOIN chinamap origin ON origin.pid = result.id ) SELECT id, name FROMresult;
