east – 第54页 – gitweixin

Spark 1月 6,2022

Spark如何在生产环境调试

接手别人开发的spark程序，大概弄懂整个流程，但一些细节总是猜不透，在生产环境运行效果也达不到理想。

想去修改，遇到下面的问题：

一、由于生产环境是运行在linux服务上的，在华为HD Insight大数据平台上，在开发机不知怎样调试。

解决方式：后来发现其实在idea是可以远程调试的：

打开工程，在菜单栏中选择“Run > Edit Configurations”。
在弹出的配置窗口中用鼠标左键单击左上角的号，在下拉菜单中选择Remote，如图1所示。图1 选择Remote

3. 选择对应要调试的源码模块路径，并配置远端调试参数Host和Port，如图2所示。
其中Host为Spark运行机器IP地址，Port为调试的端口号（确保该端口在运行机器上没被占用）。

说明：当改变Port端口号时，For JDK1.4.x对应的调试命令也跟着改变，比如Port设置为5006，对应调试命令会变更为-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=5006，这个调试命令在启动Spark程序时要用到。

4.执行以下命令，远端启动Spark运行SparkPi。 ./spark-submit –master yarn-client –driver-java-options “-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=5006” –class org.apache.spark.examples.SparkPi /opt/FI-Client/Spark2x/spark/examples/jars/spark-examples_2.11-2.1.0.jar 用户调试时需要把–class和jar包换成自己的程序，-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=5006需要换成3获取到的For JDK1.4.x对应的调试命令。

5.设置调试断点。在IDEA代码编辑窗口左侧空白处单击鼠标左键设置相应代码行断点，如图4所示，在SparkPi.scala的29行设置断点。

6.启动调试。在IDEA菜单栏中选择“Run > Debug ‘Unnamed’”开启调试窗口，接着开始SparkPi的调试，比如单步调试、查看调用栈、跟踪变量值等，如图5所示。

二、在spark executor执行的，如果看调试结果？

解决方式是在相应的rdd加上collect()方法，把结果传送到driver来看

作者 east

Java 12月 14,2021

Java多线程使用总结

Java线程的3种创建方式，使用继承方式的好处是方便传参，你可以在子类里面添加成员变量，通过set方法设置参数或者通过构造函数进行传递，而如果使用Runnable方式，则只能使用主线程里面被声明为final的变量。不好的地方是Java不支持多继承，如果继承了Thread类，那么子类不能再继承其他类，而Runable则没有这个限制。前两种方式都没办法拿到任务的返回结果，但是Futuretask方式可以。

sleep与yield方法的区别在于，当线程调用sleep方法时调用线程会被阻塞挂起指定的时间，在这期间线程调度器不会去调度该线程。而调用yield方法时，线程只是让出自己剩余的时间片，并没有被阻塞挂起，而是处于就绪状态，线程调度器下一次调度时就有可能调度到当前线程执行。

守护线程并非只有虚拟机内部可以提供，用户也可以手动将一个用户线程设定/转换为守护线程。在Thread类中提供了一个setDaemon(true)方法来将一个普通的线程（用户线程）设置为守护线

public final void setDaemon(boolean on);

如果你希望在主线程结束后JVM进程马上结束，那么在创建线程时可以将其设置为守护线程，如果你希望在主线程结束后子线程继续工作，等子线程结束后再让JVM进程结束，那么就将子线程设置为用户线程。

作者 east

Java, 数据库 12月 14,2021

华为大数据平台FusionInsight HD Redis批量删除key方法

spark程序突然跑不起来，排查后发现原来是内存满了。（可以通过redis客户端的 info memory命令查看）

./redis-cli -h 127.0.0.1 -p 6379
info memory

网上查到的批量方案

./redis-cli -h 127.0.0.1 -p 6379 keys "mykeys*" | xargs ./redis-cli -h 127.0.0.1 -p 6379 del

运行后报错。好像在华为的FusionInsight HD集群这种方案行不通。

后来通过阅读源码，在ClusterUtil的类发现可以批量删除key的方法。

  public void batchDelete(String pattern, int tryTimes)
  {
    if (tryTimes <= 0) {
      throw new IllegalArgumentException("tryTimes must be greater than or equal to 0");
    }
    ScanParams scanRarams = new ScanParams().match(pattern).count(1000);
    Set<JedisPool> pools = this.jedisCluster.getServingNodes();
    CountDownLatch latch = new CountDownLatch(pools.size() * tryTimes);
    try
    {
      for (int i = 0; i < tryTimes; i++) {
        for (JedisPool jedisPool : pools) {
          this.threadPool.submit(new DelRunnable(jedisPool, scanRarams, latch));
        }
      }
      latch.await();
    }
    catch (InterruptedException e)
    {
      throw new JedisException(e);
    }
  }

作者 east

shardingsphere 12月 10,2021

Shardingsphere 4.0.0-RC1使用遇到的坑

1、使用pagehelper后跨表查询奇慢无比

shardingsphere后面的版本还没试过，在目前版本时，结合mybatis + pagehelper组件是有存在问题的。在
Shardingsphere 4.0.0-RC1和pagehelper分页时报错解决一文解决了冲突，没想到后来又发现新的问题，在进行分表时，每个表数据几百万，查询跨几个表时，发现要等几分钟才出来结果。这样比单库查询加索引查还慢，一时很崩溃，本来分库分表为了提高速度，现在反而速度更慢了。后来在查询前把

PageHelper.startPage(pageNum, pageSize);

注释掉，刷刷一下子出来结果了。看来如果要实现分页还是自己实现。

2、查询语句太复杂，不走分片规则，进行全库查询

原来sql语句片段如下：

(VALUE1 LIKE concat(#(value,jdbcType=VARCHAR},%) AND TYPE = 1) OR  
(VALUE2 LIKE concat(#(value,jdbcType=VARCHAR},%) AND TYPE = 2) OR 
(VALUE2 LIKE concat(#(value,jdbcType=VARCHAR},%) AND TYPE = 3)

没有调用到分片规则，后来把sql改简单后就正常了。例如下面的

TYPE != 4 AND VALUE1 LIKE concat(#(value,jdbcType=VARCHAR},%) OR VALUE2 LIKE concat(#(value,jdbcType=VARCHAR},%)

作者 east

Spark 12月 6,2021

Spark Streaming多个输入流

由于业务需要，一个地方部署1个Spark Streaming程序，由于业务扩展部署了多个地方，导致大数据平台的yarn资源不足了，CPU和内存经常是100%的。而且多套只是配置不同的程序，一旦有修改，维护起来也不方便。于是想到提升Spark Streaming的并行度，同时接收多个Dstream的输入。

通过网络接收数据（如Kafka、Flume、套接字等）需要将数据反序列化并存储在Spark上，如果数据接收成为系统中的瓶颈，则需要并行接收数据。主要通过提升Receiver的并发度和调整Receiver的RDD数据分区时间隔。提升Receiver的并发度：在Worker节点上对每个输入DStream创建一个Receiver并运行，以接收一个数据流。通过创建多个输入DStream并配置从数据源接收不同分区的数据流，从而实现接收多数据流。例如，一个单Kafka输入DStream接收两个主题的数据，可以分成两个Kafka的输入流，每个仅仅接收一个主题。输入DStream运行在两个Worker节点的接收器上，从而能够并行接受并行，提高整体的吞吐量。多DStream可以通过联合（union）在一起从而创建一个DStream，这样一些应用在一个输入DStream的转换操作便可以用在联合后的DStream上。

JavaDstream<string> sources1=ssc.receiverstream(new JavacustomReceiver2(ip1, port, StorageLevel.MEMORY_ONLY-2()));

JavaDStream<String> sources2 = ssc.receiverStream(new JavaCustomReceiver2(ip2, port, StorageLevel.MEMORY_ONLY-2()));
JavaDStream<String> sources3 = ssc.receiverstream(new JavaCustomreceiver2(whip, port, StorageLeve1.MEMORY_ONLY-2()));

Javadstream<string> sources3 = ssc.socketTextstream(ip3, port, storagetevel.MEMORY ONLY2())); 
JavaDStream<String> sources = sources1.union(sources2).union(sources3);

作者 east

shardingsphere 11月 28,2021

Shardingsphere 4.0.0-RC1和pagehelper分页时报错解决

项目架构使用Shardingsphere和pagehelper架构，对一个复杂sql语句，运行后报错“Must have sharding column with subquery. "

原来 PageHelper里面有个机制是，当解析的sql比较复杂的时候，会加上别名，而Sharding-jdbc执行这个带有别名的sql会报错。如果不用pageHelper，自己来分页是可以避免这个问题。但这样做比较麻烦。

解决办法是在另加一个XXX_COUNT的sql，不要让PageHelper给原始sql加上别名。官网的做法解释如下：

增加 countSuffix count 查询后缀配置参数，该参数是针对 PageInterceptor 配置的，默认值为 _COUNT。

分页插件会优先通过当前查询的 msId + countSuffix 查找手写的分页查询。

如果存在就使用手写的 count 查询，如果不存在，仍然使用之前的方式自动创建 count 查询。

例如，如果存在下面两个查询：

<select id="selectLeftjoin" resultType="com.github.pagehelper.model.User">     select a.id,b.name,a.py from user a     left join user b on a.id = b.id     order by a.id
 </select>

 <select id="selectLeftjoin_COUNT" resultType="Long">  
   select count(distinct a.id) from user a     left join user b on a.id = b.id 
</select>

上面的 countSuffix 使用的默认值 _COUNT，分页插件会自动获取到 selectLeftjoin_COUNT 查询，这个查询需要自己保证结果数正确。

返回值的类型必须是resultType="Long"，入参使用的和 selectLeftjoin 查询相同的参数，所以在 SQL 中要按照 selectLeftjoin 的入参来使用。

因为 selectLeftjoin_COUNT 方法是自动调用的，所以不需要在接口提供相应的方法，如果需要单独调用，也可以提供。

作者 east

技术架构 11月 28,2021

阅读开源软件源码的心得体会

互联网大厂研发的职位，很多有对源码有要求。同时，看源码最大的好处是可以开阔思维，提升架构设计能力。有些东西仅靠书本和自己思考是很难学到的，必须通过看源码，看别人如何设计，然后思考为何这样设计才能领悟到。能力的提高不在于你写了多少代码，做了多少项目，而在于给你一个业务场景时，你是否能拿出几种靠谱的解决方案，并且说出各自的优缺点。而如何才能拿出来，一来靠经验，二来靠归纳总结，而看源码可以快速增加你的经验。而不少源码十分庞大复杂，下面谈谈阅读源码心得体会。

那么如何阅读源码呢？在你看某一个框架的源码前，先去Google查找这个开源框架的官方介绍，通过资料了解该框架有几个模块，各个模块是做什么的，之间有什么联系，每个模块都有哪些核心类，在阅读源码时可以着重看这些类。或者找找是否有这方面源码解读的书，在别人探索好的路再去探索，能节省时间。

然后对哪个模块感兴趣就去写个小demo，先了解一下这个模块的具体作用，然后再debug进入看具体实现。在debug的过程中，第一遍是走马观花，简略看一下调用逻辑，都用了哪些类；第二遍需有重点地debug，看看这些类担任了架构图里的哪些功能，使用了哪些设计模式。如果第二遍有感觉了，便大致知道了整体代码的功能实现，但是对整体代码结构还不是很清晰，毕竟代码里面多个类来回调用，很容易遗忘当前断点的来处；那么你可以进行第三遍debug，这时候你最好把主要类的调用时序图以及类图结构画出来，等画好后，再对着时序图分析调用流程，就可以清楚地知道类之间的调用关系，而通过类图可以知道类的功能以及它们相互之间的依赖关系。另外，开源框架里面每个功能类或者方法一般都有注释，这些注释是一手资料，比如JUC包里的一些并发组件的注释，就已经说明了它们的设计原理和使用场景。

在阅读源码时，最好画出时序图和类图，因为人总是善忘的。如果隔一段时间你再去看之前看过的源码，虽然有些印象，但当你想去看某个模块的逻辑时，又需根据demo再从头debug了。而如果有了这俩图，就可以从这俩图里面直接找，并且看一眼时序图就知道整个模块的脉络了。

作者 east

Java 11月 25,2021

优化后台接口经验总结

开发一个后台接口不难，可能花几个小时就能跑通；而做一个好用的接口，可能要花几天时间精雕细磨。尤其是业务复杂，数据量大的，如果没有优化导致速度很慢，用户根本没耐心等。

下面谈谈优化思路：

1、串行操作改为并发操作。如果接口有串行操作，而且其中一些操作是比较耗时的，而且它们的操作没有因果关系需要等待前面的结果，那么可以把这些操作改为并发线程去操作。使用多线程遇到坑会多一些，例如线程池使用FutureTask时如果把拒绝策略设置为DiscardPolicy和DiscardOldestPolicy，并且在被拒绝的任务的Future对象上调用了无参get方法，那么调用线程会一直被阻塞。在日常开发中尽量使用带超时参数的get方法以避免线程一直阻塞。

2、对一些常用的可复用的数据库查询加上缓存。做一个系统需要经常查询点位信息，这些点位信息不会经常变，对相关的数据库查询加上缓存。如果用mybatis，可以考虑直接在xml上添加，或者利用redis进行添加。

3、数据的优化，如果单表数据过大，可以考虑进行分库分表。

4、对查询慢的语句，考虑加上相关的索引。频繁单个查询或插入，考虑是否能改为批量操作。对一些频繁查询可复用的，可以考虑一次批量查询出来并缓存起来。

5、考虑是否可以后台先计算出中间结果或最终结果，用户查询时不用从头开始计算。在接口查询计算过程，如果有频繁重复计算的，可以考虑采用备忘录算法。

作者 east

数据库 11月 23,2021

对接第三方数据库的数据遇到的坑

对接第三方的数据，根据轮询他们的数据库来对接数据。看到表设计有create_time字段，根据经验主义觉得是写入数据库的时间。于是想到对接数据方案是：如果查询时间小于当前时间，每几分钟查一次。如果查询时间大于当前时间，休眠到查询结束时间等于当前时间。后来发现一个奇怪现象：如果刚运行程序补录数据，发现没有漏数据，如果跑一段时间，追上当前时间，就出现漏数据。由于是采用jdbc框架的，不是很清楚底层，当时怀疑会不会运行久了断开数据库连接。反复修改程序还是出现这种情况，后来问第三方厂家，他们说create_time是服务接收到数据的时间，还要先写临时库，再写目标库。并且查询到第三方的数据库时间是落后标准时间的。

作者 east

Java, 运维 11月 7,2021

一次诡异系统变慢排查

给客户上线了系统（Spring Cloud、微服务，centos上运行），运行1、2年后，客户投诉系统很慢。

自己打开系统，刚开始很快，用一些时间就变慢。看前端请求的接口，是挂起状态，有的要几分钟才有结果。

检查服务器内存和CPU。这是引起系统慢常见的问题，发现这2方面改善后还是变慢，后来干脆重启服务器，依然无解。

检查磁盘坏道。之前用电脑时，如果磁盘有坏道，如果有写操作，有时也会为坏道。用centos检查坏道的命令也没发现。

检查接口代理。由于是前后端分离，用户在前端的请求，都经过第三方代理。如果直接测后台接口是很快，但通过前端访问就变慢了，于是怀疑是第三方代理搞的鬼。于是咨询第三方代理，第三方代理说他们服务的客户，都没有发现这种情况，建议我们排查网络。

检查前端。由于后来又上线几个类似的系统，前端基本一样的，没有安装新的插件。所以觉得可能性不大。

检查浏览器。网上有的说是chrome浏览器早期的bug，如果是已经请求过的接口，会复用之前的。想找不同浏览器或更新到最新chrome的。但几个一样的系统，用同样浏览器也没有变慢，也解释不通。

终极答案。经反复排查，最后发现是前端有个页面每几秒钟请求1次，而请求相关的数据库年长月久数据很多，数据库不能及时响应请求。后来根据业务需要改成几分钟请求1次，果然这个诡异问题没再出现。

作者 east

数据库 9月 4,2021

利用 Navicat 事件设计器解决多微服务对表不同的需求

在生产实践有一个项目，某个表每天新增数据几百万，甚至上千万条。项目采用多服务，有的同事的微服务只需要调用这个表几天的数据，有的同事只需要最近30天，而我这边的微服务需要查询这个表任意一段时间。

由于mysql访问数据超过上千万的数据性能下降很多，而且有的功能需要一些统计，加上原来项目是用mysql，考虑开发成本，暂时不考虑换乘nosql的方案，例如es、hbase等。于是改造成用shardingsphere每月分表。然而问题产生了，我这边访问的表名是TableA_2021_9这个的，而别的几位同事需要访问TableA。

后来想了一个解决方案，不需要他们也改造代码。方案就是保留原来的TableA，采用navicat事件进行调度，TableA只保留最近30天的数据，并每天运行定时任务，把新产生的数据复制一份到分月表。

INSERT INNORE INTO tableA_2021_10 SELECT FROM TableA WHERE START_TIME > DATE_ADD (NOW(), INTERVAL -1 DAY) AND START_TIME < NOW()

作者 east

数据库 8月 28,2021

mysql数据去重实践总结

在线上服务器，由于程序的bug，导入很多重复的数据。刚开始想到的思路是直接写SQL进行去重。例始要对login_name这个字段进行去重：

DELETE
FROM
    `oldTable`
WHERE
    login_name IN (
        SELECT
            a.login_name
        FROM
            (
                SELECT
                    login_name
                FROM
                    `oldTable`
                GROUP BY
                    login_name
                HAVING
                    count(login_name) > 1
            ) AS a
    )
AND id NOT IN (
    SELECT
        b.aa
    FROM
        (
            SELECT
                min(id) AS aa
            FROM
                `oldTable`
            GROUP BY
                login_name
            HAVING
                count(login_name) > 1
        ) AS b
);

如果mysql数量一多，用上面的方法操作效率是很低的。这时要用mysql联合索引，可以建立一个表跟oldTable结构一样的表，并对login_name建立联合索引。

alter table newTableadd unique index(login_name);

如果数择量很大，可以根据时间或主键id值进行分批插入。

INSERT IGNORE INTO newTable SELECT * FROM oldTable WHERE START_TIME >= '2021-08-20 00:00:00' AND START_TIME <'2021-08-23 00:00:00'

作者 east