大数据开发 – 第27页

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

Java 11月 25,2021

优化后台接口经验总结

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

下面谈谈优化思路：

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

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

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

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

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

作者 east

Java, 运维 11月 7,2021

一次诡异系统变慢排查

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

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

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

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

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

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

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

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

作者 east

flume 7月 10,2021

Flume对接数据遇到的坑

业务是这样的：别的地方服务器通过ftp传来一些压缩包，对压缩包进行解压，然后flume进行采集发到kafka，spark Streaming进行处理。

进行解压的脚本代码如下:

#/bin/bash
end = "${dirname "$0"}"/"$1"
source = "${dirname "$0"}"/"$2"
final =  "${dirname "$0"}"/"$3"
while [2 -gt 1 ]
do
   for i in 'ls ${source}‘
   do
   if [[ ${i} != "" && ${i} != *.tmp ]]
   then
       unzip ${source}/${i} -d ${end}/
       mv -f ${source}/${i} ${final}
   fi
done
sleep 5

原本一直持续能解压文件，最近出现停止。

执行命令时，还出现提示是否覆盖。这时才明白可能是这个原因导致脚本没能顺利执行。为了不提示是否覆盖，可以加参数 -o。修改命令如下：

unzip -o ${source}/${i} -d ${end}/

作者 east

bug清单, flume 6月 13,2021

Flume pollDelay设置不正确停止采集

使用FusionInsight HD Flume从本地采集静态日志（ Spooling Source ）保存到Kafka，由于采集堆积太多了，flume配置参数做了一些修改。后来发现一个诡异问题：每次重启flume采集，只采集1、2个文件就停止采集了，也没报什么错误。

采用对比法排查问题，对比正常运行的flume配置，看到pollDelay跟之前的不同。才想起之前一顿三百五的操作：想加快速度。pollDelay的设置值从5000改成500。

采集方案采用的Spooling Source + Memory Channel + kfaka

Spooling Source常用配置：

Memory Channel使用内存作为缓存区，Events存放在内存队列中。常用配置如下表所示：

Kafka Sink将数据写入到Kafka中。常用配置如下表所示：

参考配置如下：

a1.channels = c1
a1.sources = s1
a1.sinks = sink1

a1.sources.s1.type = spooldir
a1.sources.s1.channels = c1
a1.sources.s1.spoolDir = /home/ftp （填写实际的路径）
a1.sources.s1.bufferMaxLineLength = 1073741824
a1.sources.s1.pollDelay = 5000
a1.sources.s1.consumeOrder = random

a1.channels.c1.type = memory
a1.channels.c1.capacity = 30000
a1.channels.c1.tansactionCapacity = 30000

a1.sinks.sink1.channel = c1
a1.sinks.sink1.type = org.apache.kafka.kafkaSink
a1.sinks.sink1.bootstrap.servers=192.168.1.1:210007  (根据实际填写）
a1.sinks.sink1.topic = mytopic （根据实际填写）
a1.sinks.sink1.batchSize = 200
a1.sinks.sink1.producer.requiredAcks = 1

作者 east

bug清单, Kafka, Spark 6月 10,2021

运行 Spark Streaming出现”Could not find KafkaClient entry in the JAAS configuration”

在使用FusionInsight HD大数据平台，用Spark Streaming来处理数据接入，kafka作为消费者，运行程序时出现”Could not find KafkaClient entry in the JAAS configuration”，当时怀疑是FusionInsight HD 的客户端相关配置有问题。

采用替换法思维，在另一台已经验证 FusionInsight HD 的客户端没问题的服务上运行，果然这个问题没做出现，仔细对比了这2台服务FusionInsight HD 的客户端的配置，发现在spark配置文件（hd安装目录/Spark2x/spark/conf/Jaas.conf）要修改为下面的配置：

KafkaClient{
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
principal="大数据平台账号"
useTicketCache=false
keyTab="user.keytab的路径"
storeKey=true;
};

作者 east

Hbase 6月 4,2021

使用Hbase出现KeeperErrorCode = Session expired for /hbase/hbaseid

在使用FusionInsight HD大数据平台安全认证模式时，用hbase Shell或代码调用hbase时，有时出现”KeeperErrorCode = Session expired for /hbase/hbaseid”等错误。这是因为安全认证session过时了。

可以切换到hd，使用HBase客户端安装目录，输入下面命令

source bigdata_env
kinit 组件业务用户
按提示输入密码

再使用hbase shell就不会上面的问题。
如果是在Spark上调用的话，可以在调用的shell脚本上kinit 指定机机认证的安全认证文件。或者干脆写个定时执行的脚本配置在crontab上。

作者 east

neo4j 5月 27,2021

Neo4j的地理空间特性

很多数据库都有地理空间特性，例如mysql在5.0以上就有，es也有。

从Neo4j 3.0版本开始就内置了有限的空间支持功能。默认支持点和距离。假定点设置了经纬度属性值，就可计算出两点之间的距离。先让我们创建塔，并养成好习惯定义相应的约束：

CREATE CONSTRAINT ON （t:Tower) ASSERT t.name IS UNIQUE
CREATE (paris:Tower {name:"Eiffel Tower",country:"FRA",latitude:48.82322,longitude:2.29323})

CREATE (guangzhou:Tower {name:"Guangzhou Tower",country:"CN",latitude:68.82322,longitude:128.29323})

可通过下面查询语句获得埃菲尔铁塔和其他塔之间的距离：

MATCH (et:Tower {name:'Eiffle Tower'}), (other:Tower)
RETURN et.name, other.name, round(distance(point(et),
point(other))/10)/100 as dist_km

这里的关键是点（point），一个地理位置点是用图数据库中的一个带有纬度和经度属性的节点进行表示，distance函数是通过两点位置计算出来的。Round函数的作用是舍入到最近的整数，使用小技巧（/10/100），可将数值保留到小数点后两位

可通过以下方式查询APOC的空间过程列表：

CALL apoc.help("spatial")

APOC提供了地理编码的可能性，它将地址转换成空间坐标并按距离对路径进行排序。虽然它只是将地址映射到坐标，但不只是两个浮点数，地理编码并不容易，大量工程师在通过代码进行地理编码的变换。存储汽车行走时拍摄的照片需要消耗大量存储资源，识别图片中的数字需要消耗大量的CPU计算资源。因此，首选是不要自己来处理地理编码，而是依赖一个可被APOC透明调用的提供方。

作者 east

Java 5月 13,2021

Java远程访问工具类

我们都知道，xshell、xftp等是远程访问的利器，但在实际工作中，有时要用java代码实现上面这2个工具的基础功能，远程访问和下载等等：

public class RemoteShellUtil {

    private final static Logger LOGGER = LoggerFactory.getLogger(RemoteShellUtil.class);

    private JSch jsch;

    private Session session = null;

    private String user = null;

    private String password = null;

    private String host = null;

    public void setUser(String user) {
        this.user = user;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    public void setHost(String host) {
        this.host = host;
    }

    public void exec(String command) {
        if (this.session == null || !this.session.isConnected()) {
            if (!this.Connect(this.user, this.password, this.host)) {
                LOGGER.error("connect failed");
                return;
            }
        }

        BufferedReader reader = null;
        Channel channel = null;

        try {
            channel = this.session.openChannel("exec");
            ((ChannelExec) channel).setCommand(command);
            channel.setInputStream(null);
            ((ChannelExec) channel).setErrStream(System.err);
            channel.setInputStream(null);
            int exitStatus = channel.getExitStatus();
            channel.connect();
            InputStream in = channel.getInputStream();
            reader = new BufferedReader(new InputStreamReader(in));
            String buf;

            while ((buf = reader.readLine()) != null) {
                LOGGER.info(buf);
            }
        } catch (JSchException | IOException e) {
            LOGGER.error(Throwables.getStackTraceAsString(e));
        } finally {
            try {
                if (null != reader) reader.close();
                if (null != channel) channel.disconnect();
                this.session.disconnect();
            } catch (IOException e) {
                LOGGER.error(Throwables.getStackTraceAsString(e));
            }

        }

    }

    public Boolean download(String serverPath, String localpath) {
        if (!checkConnect()) return false;

        ChannelSftp channel = null;

        boolean status = false;
        try {
            channel = (ChannelSftp) this.session.openChannel("sftp");
            channel.connect();
            FileOutputStream out = new FileOutputStream(new File(localpath));
            BufferedOutputStream writer = new BufferedOutputStream(out);
            SftpProgressMonitor var10000 = new SftpProgressMonitor() {
                private long current = 0L;

                @Override
                public void init(int i, String s, String s1, long l) {
                }

                public boolean count(long arg0) {
                    this.current += arg0;
                    return true;
                }

                @Override
                public void end() {
                }
            };
            channel.get(serverPath, writer);
            writer.close();
            status = true;
        } catch (Exception e) {
            LOGGER.error(Throwables.getStackTraceAsString(e));
        } finally {
            if (null != channel) channel.disconnect();
            this.session.disconnect();
        }

        return status;
    }

    public Boolean upload(String serverPath, String localpath) {
        if (!checkConnect()) return false;

        ChannelSftp channel = null;

        boolean status = false;
        try {
            channel = (ChannelSftp) this.session.openChannel("sftp");
            channel.connect();
            FileInputStream in = new FileInputStream(new File(localpath));
            BufferedInputStream reader = new BufferedInputStream(in);
            SftpProgressMonitor var10000 = new SftpProgressMonitor() {
                private long current = 0L;

                public boolean count(long arg0) {
                    this.current += arg0;
                    return true;
                }

                public void end() {
                }

                public void init(int arg0, String arg1, String arg2, long arg3) {
                }
            };
            channel.put(reader, serverPath);
            reader.close();
            status = true;
        } catch (Exception e) {
            LOGGER.error(Throwables.getStackTraceAsString(e));
        } finally {
            if (null != channel) channel.disconnect();
            this.session.disconnect();
        }

        return status;
    }

    private Boolean Connect(String user, String password, String host) {
        try {
            if (null == this.jsch) this.jsch = new JSch();
            this.session = this.jsch.getSession(user, host, 22);
            this.session.setPassword(password);
            Properties config = new Properties();
            config.put("StrictHostKeyChecking", "no");
            this.session.setConfig(config);
            this.session.connect();
            return true;
        } catch (Exception e) {
            LOGGER.error("connect failed, reason{}", Throwables.getStackTraceAsString(e));
            return false;
        }
    }

    private Boolean checkConnect() {
        if (this.session == null || !this.session.isConnected()) {
            return this.Connect(this.user, this.password, this.host);
        }
        return true;
    }
}

作者 east

分类归档大数据开发