east – 第66页 – gitweixin

python 2月 21,2021

怎样在3分钟内安装PySpark 和 Jupyter Notebook

对于大数据爱好者来说，Apache Spark是必须的。简而言之，Spark是一个快速而强大的框架，它提供了一个API，可以对弹性数据集执行大规模的分布式处理。 Jupyter Notebook是一个流行的应用程序，它使您可以编辑，运行并将Python代码共享到Web视图中。它允许您以非常灵活的方式修改和重新执行部分代码。因此，Jupyter是测试程序和原型的好工具。 Jupyter Notebook运行Python代码 Jupyter Notebook运行Python代码我为Linux用户撰写了这篇文章，但是我确信Mac OS用户也可以从中受益。为什么在Jupyter Notebook中使用PySpark？使用Spark时，大多数数据工程师建议使用Scala（这是“本机” Spark语言）或通过完整的PySpark API使用Python开发。 Python for Spark显然比Scala慢。但是，像许多开发人员一样，我喜欢Python，因为Python灵活，健壮，易学并且可以从我所有的收藏夹库中受益。我认为，Python是大数据/机器学习领域中进行原型开发的理想语言。现在，让我们开始吧。

安装pySpark

在安装pySpark之前，必须先安装Python和Spark。我在以下示例中使用的是Python 3，但您可以轻松地使其适应Python2。请访问Python官方网站进行安装。我也鼓励您建立一个virtualenv
要安装Spark，请确保您的计算机上安装了Java 8或更高版本。然后，访问Spark下载页面。选择最新的Spark版本（一个针对Hadoop的预构建包），然后直接下载。

解压缩并将其移动到您的/ opt文件夹：

$ tar -xzf spark-1.2.0-bin-hadoop2.4.tgz

$ mv spark-1.2.0-bin-hadoop2.4 /opt/spark-1.2.0

创建一个软连接

$ ln -s /opt/spark-1.2.0 /opt/spark̀

这样，您将能够下载和使用多个Spark版本。最后，告诉您的bash（或zsh等）在哪里可以找到Spark。为此，通过在〜/ .bashrc（或〜/ .zshrc）文件中添加以下几行来配置$ PATH变量：

安装Jupyter Notebook

$ pip install jupyter
$ jupyter notebook

您在Spark上的第一个Python程序
让我们检查是否在没有先使用Jupyter Notebook的情况下正确安装了PySpark。

您可能需要重新启动终端才能运行PySpark

import random
num_samples = 100000000

def inside(p):     
  x, y = random.random(), random.random()
  return x*x + y*y < 1

count = sc.parallelize(range(0, num_samples)).filter(inside).count()

pi = 4 * count / num_samples
print(pi)

sc.stop()

Jupyter使用PySpark

有两种方法可以在Jupyter Notebook中使用PySpark：

配置PySpark驱动程序以使用Jupyter Notebook：运行pyspark将自动打开Jupyter Notebook

加载常规的Jupyter Notebook并使用findSpark软件包加载PySpark

第一种选择更快，但特定于Jupyter Notebook，第二种选择是一种更广泛的方法，可以在您喜欢的IDE中使用PySpark。

方法1 —配置PySpark驱动程序
更新PySpark驱动程序环境变量：将这些行添加到〜/ .bashrc（或〜/ .zshrc）文件中。

export PYSPARK_DRIVER_PYTHON=jupyter
export PYSPARK_DRIVER_PYTHON_OPTS='notebook'

重启并加载：

$ pyspark

现在，此命令应在您的Web浏览器中启动Jupyter Notebook。通过单击“新建”>“ Notebooks Python [默认]”来创建一个新笔记本。

复制并粘贴我们的Pi计算脚本，然后按Shift + Enter运行它。

方法2 — FindSpark程序包
在Jupyter Notebook中使用PySpark还有另一种更通用的方法：使用findSpark包在代码中提供Spark上下文。

findSpark包并非特定于Jupyter Notebook，您也可以在自己喜欢的IDE中使用此技巧。

要安装findspark：

$ pip install findspark
$ jupyter notebook

创建一个新的Python [default]笔记本并编写以下脚本：

import findspark
findspark.init()

import pyspark
import random

sc = pyspark.SparkContext(appName="Pi")
num_samples = 100000000

def inside(p):     
  x, y = random.random(), random.random()
  return x*x + y*y < 1

count = sc.parallelize(range(0, num_samples)).filter(inside).count()

pi = 4 * count / num_samples
print(pi)

sc.stop()

输出结果：

作者 east

Kafka, Spark 2月 21,2021

spark和kafka在数据流处理对比

2625 / 5000

在对Spark Streaming和Kafka Streaming进行比较并得出何时使用哪个比较之前，让我们首先对Data Streaming的基础知识有一个清晰的了解：它是如何出现的，流是什么，如何运行，其协议和用例。。数据流如何诞生？从那时起，数据一直是操作的重要组成部分。数据构成了整个操作结构的基础，其中数据被进一步处理以在系统的不同实体模块中使用。这就是为什么它已成为IT领域的典型代表。随着技术的发展，数据的重要性变得更加突出。数据处理中使用的方法已经发生了显着变化，以适应软件机构对数据输入的不断增长的需求。随着时间的增长，数据处理的时间框架急剧缩短，以至于立即处理的输出有望满足最终用户的更高期望。随着人工智能的出现，人们强烈希望为看起来像人类的最终用户提供实时帮助。此要求仅取决于数据处理强度。越快越好。因此，结果是处理数据的方式发生了变化。较早之前，在指定的延迟之后，有成批的输入被输入到系统中，从而将处理后的数据作为输出。目前，这种延迟（延迟）是输入性能，处理时间和输出的结果，这已成为性能的主要标准之一。为了确保高性能，延迟必须最小化到几乎是实时的程度。这就是数据流出现的方式。在数据流处理中，实时数据流作为输入传递，必须立即进行处理，并实时传递输出信息流。

什么是数据流？

数据流传输是一种方法，其中不按常规的批处理方式发送输入，而是以连续流的形式发布该流，并按原样使用算法进行处理。还以连续数据流的形式检索输出。该数据流是使用数千个源生成的，这些源同时以小尺寸发送数据。这些文件背对背发送时形成连续的流程。这些可能是大量发送的日志文件以进行处理。这种作为流出现的数据必须被顺序处理以满足（几乎）连续实时数据处理的要求。

为什么需要数据流？

随着企业在线人数的增加以及随之而来的对数据的依赖，人们已经意识到了数据的方式。数据科学和分析技术的出现导致大量数据的处理，为实时数据分析，复杂数据分析，实时流分析和事件处理提供了可能性。

当输入数据大小庞大时，需要进行数据流传输。我们需要先存储数据，然后再将其移动以进行批处理。由于数据以多批次的形式存储，因此涉及大量时间和基础架构。为了避免所有这些情况，信息以小数据包的形式连续流传输以进行处理。数据流提供超可伸缩性，这仍然是批处理的挑战。

使用数据流传输的另一个原因是要提供近乎实时的体验，其中最终用户在输入数据时会在几秒钟或几毫秒内获得输出流。

当数据源似乎无穷无尽且无法为批处理中断时，也需要进行数据流传输。 IoT传感器在此类别中发挥了重要作用，因为它们会生成连续的读数，需要对其进行处理以得出推论。

数据流如何发生？

为了通过实时处理数据做出即时决策，可以进行数据流传输。根据系统的规模，复杂性，容错性和可靠性要求，您可以使用工具，也可以自己构建。

自行构建它意味着您需要在编码角色之前将事件放置在诸如Kafka之类的消息代理主题中。这里的参与者是一段代码，旨在接收来自代理中的问题的事件（即数据流），然后将输出发布回代理。

Spark是第一代Streaming Engine，它要求用户编写代码并将其放置在actor中，他们可以进一步将这些actor连接在一起。为了避免这种情况，人们经常使用Streaming SQL进行查询，因为它使用户可以轻松地查询数据而无需编写代码。流SQL是对SQL的扩展支持，可以运行流数据。此外，由于SQL在数据库专业人员中已得到很好的实践，因此执行流式SQL查询将更加容易，因为它基于SQL。

这是用例的流式SQL代码，在这种情况下，如果池中的温度在2分钟内下降了7度，则必须向用户发送警报邮件。

@App:name("Low Pool Temperature Alert")

@App: description('An application which detects an abnormal decrease in swimming pools temperature.')

@source(type='kafka',@map(type='json'),bootstrap.servers='localhost:9092',topic.list='inputStream',group.id='option_value',threading.option='single.thread')

define stream PoolTemperatureStream(pool string, temperature double);

@sink(type='email', @map(type='text'), ssl.enable='true',auth='true',content.type='text/html', username='sender.account', address='sender.account@gmail.com',password='account.password', subject="Low Pool Temperature Alert", to="receiver.account@gmail.com")

define stream EmailAlertStream(roomNo string, initialTemperature double, finalTemperature double);

--Capture a pattern where the temperature of a pool decreases by 7 degrees within 2 minutes

@info(name='query1')

from every( e1 = PoolTemperatureStream ) -> e2 = PoolTemperatureStream [e1.pool == pool and (e1.temperature + 7.0) >= temperature]

    within 2 min

select e1.pool, e1.temperature as initialTemperature, e2.temperature as finalTemperature

insert into EmailAlertStream;

Spark SQL提供DSL（特定于域的语言），这将有助于以不同的编程语言（例如Scala，Java，R和Python）操纵DataFrame。它使您可以使用SQL或DataFrame API对Spark程序内部的结构化数据执行查询。 Kafka等新一代流引擎也支持Kafka SQL或KSQL形式的Streaming SQL。

尽管流处理的过程大致相同，但此处重要的是根据用例要求和可用的基础结构选择流引擎。在得出结论之前，什么时候使用Spark Streaming和什么时候使用Kafka Streaming，让我们首先探索Spark Streaming和Kafka Streaming的基础知识，以更好地理解。

什么是Spark Streaming？

Spark Streaming是核心Spark API的扩展，可让其用户执行实时数据流的流处理。它从Kafka，Flume，Kinesis或TCP套接字等来源获取数据。可以使用复杂的算法对这些数据进行进一步处理，这些复杂的算法使用诸如map，reduce，join和window之类的高级功能表示。最终输出（即处理后的数据）可以推送到诸如HDFS文件系统，数据库和实时仪表板之类的目标。

让我们仔细看看Spark Streaming的工作原理。 Spark Streaming从数据源以数据流的形式获取实时输入，并将其进一步分为几批，然后由Spark引擎处理以生成大量输出。 Spark Streaming允许您将机器学习和图形处理用于数据流以进行高级数据处理。它还提供了代表连续数据流的高级抽象。数据流的这种抽象称为离散流或DStream。该DStream可以通过对Kafka，Flume和Kinesis等来源的数据流或其他DStream进行高级操作来创建。这些DStream是RDD（弹性分布式数据集）的序列，RDD是分布在计算机集群上的多个只读数据集。这些RDD以容错方式进行维护，使其具有高度鲁棒性和可靠性。DStreams序列Spark Streaming使用Spark Core的快速数据调度功能来执行流分析。从诸如Kafka，Flume，Kinesis等之类的源中以迷你批的形式摄取的数据用于执行数据流处理所需的RDD转换。

Spark Streaming使您可以根据需要使用Scala，Java或Python编写程序来处理数据流（DStreams）。由于此处将用于批处理的代码用于流处理，因此使用Spark Streaming实现Lambda体系结构（将批处理和流处理混合在一起）变得容易得多。但这是以等于最小批处理持续时间的延迟为代价的。 Spark Streaming中的输入源 Spark支持主要来源，例如文件系统和套接字连接。另一方面，它也支持高级资源，例如Kafka，Flume，Kinesis。只有添加额外的实用程序类，才能获得这些出色的资源。您可以使用以下工件链接Kafka，Flume和Kinesis。

kafka：spark-streaming-kafka-0-10_2.12

flume：spark-streaming-flume_2.12

Kinesis：spark-streaming-kinesis-asl_2.12

什么是Kafka流媒体？

Kafka Stream是一个客户端库，可让您处理和分析从Kafka接收的数据输入，并将输出发送到Kafka或其他指定的外部系统。 Kafka依赖于流处理概念，例如：准确区分事件时间和处理时间窗口支持高效直接的应用程序状态管理通过利用Kafka中的生产者和消费者库来利用Kafka的本机功能，从而简化了应用程序开发，从而使其更加直接和快捷。正是由于这种原生的Kafka潜力，使得Kafka流式传输可以提供数据并行性，分布式协调，容错性和操作简便性。 Kafka Streaming中的主要API是提供多个高级运算符的流处理DSL（特定于域的语言）。这些运算符包括：筛选器，映射，分组，窗口，聚合，联接和表的概念。 Kafka中的消息传递层对进一步存储和传输的数据进行分区。根据状态事件在Kafka流中对数据进行分区，以进行进一步处理。通过将拓扑划分为多个任务来缩放拓扑，其中为每个任务分配了输入流中的分区列表（Kafka主题），从而提供了并行性和容错能力。

Kafka可以进行状态转换，与Spark Streaming中的批处理不同。它在其主题内存储状态，流处理应用程序将其用于存储和查询数据。因此，其所有操作均受状态控制。这些状态还用于连接主题以形成事件任务.Kafka中基于状态的操作这是由于Kafka中基于状态的操作使其具有容错能力，并允许从本地状态存储中自动恢复。 Kafka Streaming中的数据流是使用表和KStreams的概念构建的，这有助于它们提供事件时间处理。

Spark Streaming与Kafka Streaming：

何时使用什么 Spark Streaming使您可以灵活地选择任何类型的系统，包括具有lambda架构的系统。但是，Spark Streaming的延迟范围从毫秒到几秒。如果延迟不是一个重要的问题，并且您正在寻找在源兼容性方面的灵活性，那么Spark Streaming是最佳选择。可以在EC2，Hadoop YARN，Mesos或Kubernetes上使用独立的集群模式运行Spark Streaming。它可以访问HDFS，Alluxio，Apache Cassandra，Apache HBase，Apache Hive和许多其他数据源中的数据。它提供了容错能力，还提供了Hadoop分发。此外，在Spark流式传输的情况下，您不必为批处理和流式传输应用程序分别编写多个代码，在这种情况下，单个系统可以同时满足这两种情况。另一方面，如果延迟是一个重要问题，并且必须坚持以短于毫秒的时间范围进行实时处理，则必须考虑使用Kafka Streaming。由于事件驱动处理，Kafka Streaming提供了高级的容错能力，但是与其他类型的系统的兼容性仍然是一个重要的问题。此外，在高可伸缩性要求的情况下，Kafka具有最佳的可伸缩性，因此非常适合。

如果您要处理从Kafka到Kafka的本机应用程序（输入和输出数据源都在Kafka中），则Kafka流式传输是您的理想选择。虽然Kafka Streaming仅在Scala和Java中可用，但Spark Streaming代码可以用Scala，Python和Java编写。结束语随着技术的发展，数据也随着时间大量增长。处理此类海量数据的需求以及对实时数据处理的日益增长的需求导致了数据流的使用。通过几种数据流方法，尤其是Spark Streaming和Kafka Streaming，全面了解用例以做出最适合需求的最佳选择变得至关重要。在用例中优先考虑需求对于选择最合适的流技术至关重要。鉴于事实，Spark Streaming和Kafka Streaming都是高度可靠的，并且广泛推荐作为Streaming方法，它在很大程度上取决于用例和应用程序，以确保最佳效果。在本文中，我们指出了两种流传输方法的专业领域，以便为您提供更好的分类，这可以帮助您确定优先级并做出更好的决策。

作者 east

flume 2月 20,2021

Flume处理Spooling Directory Source数据太慢优化

一个数据采集处理系统，架构如下：

日志数据 -> flume -> kafka -> Spark Streaming。

flume到kafka的数据处理时间是1秒多；而spark Streaming的数据处理时间是十几毫秒。

flume方面：

flume数量不够：增加日志服务器以增加并行度；

（1）自身：增加内存flume-env.sh 4-6g

-Xmx与-Xms最好设置一致，减少内存抖动带来的性能影响，如果设置不一致容易导致频繁fullgc。

（2）找朋友：多个目录，多个spooling directory source同时采集

（3）taildir source的batchsize修改为合适的值

( 4 ) flume要读取文件夹如果文件太多，要按最新或最早顺序读取时，会很影响速度。

# batchsize是每次处理的数据条数越高，处理的数据越多，延迟越高。

kafka数据积压的问题，主要的解决办法是：

（1）增加Kafka对应的分区数(比如：期望处理数据的总吞吐量是100M/s。但是实际最多每个分区的生产能力和消费能力的最小值是20M/s，那么我们就需要设置5个或者6个分区)，2）要求下一级消费者配套增加CPU核数，动态增加Kafka服务器集群。

（2）kafka ack设成0(ack有0有1有-1。0的可靠性最差，但是速度最快)

注：ack有3个可选值，分别是1，0，-1。

ack=1，简单来说就是，producer只要收到一个分区副本成功写入的通知就认为推送消息成功了。这个副本必须是leader副本。只有leader副本成功写入了，producer才会认为消息发送成功。

ack=0，简单来说就是，producer发送一次就不再发送了，不管是否发送成功。

ack=-1，简单来说就是，producer只有收到分区内所有副本的成功写入的通知才认为推送消息成功了。

（3） a1.channels.c1.type = memory memory类型可能会丢失数据，但是速度最快。

作者 east

bug清单, flume 2月 20,2021

Flume Spooling Directory Source 采集NullPointerException

采用flume的Spooling Directory Source，突然遇到下面的错误

(org.apache.flume.source.SpoolDirectoryExtSource2$SpoolDirectoryRunnable.run:277)  - FATAL: Spool Directory source source1: { spoolDir: /home/work/local/log }: Uncaught exception in SpoolDirectorySource thread. Restart or reconfigure Flume to continue processing.
java.lang.NullPointerException
    at org.apache.avro.io.ResolvingDecoder.readLong(ResolvingDecoder.java:159)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:157)
    at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:177)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:148)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:139)
    at org.apache.avro.file.DataFileStream.next(DataFileStream.java:233)
    at org.apache.flume.serialization.DurablePositionTracker.initReader(DurablePositionTracker.java:171)
    at org.apache.flume.serialization.DurablePositionTracker.<init>(DurablePositionTracker.java:158)
    at org.apache.flume.serialization.DurablePositionTracker.getInstance(DurablePositionTracker.java:76)
    at org.apache.flume.client.avro.ReliableSpoolingFileEventExtReader2.openFile(ReliableSpoolingFileEventExtReader2.java:561)
    at org.apache.flume.client.avro.ReliableSpoolingFileEventExtReader2.getNextFile(ReliableSpoolingFileEventExtReader2.java:511)
    at org.apache.flume.client.avro.ReliableSpoolingFileEventExtReader2.readEvents(ReliableSpoolingFileEventExtReader2.java:264)
    at org.apache.flume.source.SpoolDirectoryExtSource2$SpoolDirectoryRunnable.run(SpoolDirectoryExtSource2.java:252)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
    at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:304)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:178)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)

重启flume 后发现问题依旧。试验把flume 采集的日志放到新目录，重启flume采集新目录发现没问题了。看来是旧目录/home/work/local/log 有文件变动造成的。

例如有.flumespool隐藏目录，该目录下有文件.flumespool-main.meta隐藏文件，用来记录flume读取文件的位置，发现该记录停止在flume出问题的时间。再查看其它正常运行的机器的相同路径及文件，并没有发现该文件，于是将该文件移到其它目录下，重启flume，此时发现flume成功运行！

作者 east

运维 2月 19,2021

linode如何迁移服务器

linode是个性价比的服务器，广受广大站长喜欢。由于种种原因，可以很方便进行不同地区迁移。例如linode日本地区服务器，在中国用移动网络不太稳定，要迁移到加州。

方法/步骤

虽然迁移服务器理论上讲只是ip变了，不影响服务上网站运行。但为防止万一，最后进行备份。登录linode服务器，找到要迁移的服务器，选择”Migrate”
在下一步的界面中，勾选同意，选择要迁移的服务器地区。最后点击“Enter Migrate Queue”
完成上一步后，可以在服务器的status上看到迁移的进度。迁移完成后会自动重启。
迁移服务后，由于ip变了，所以要在域名解析中，把新ip替换旧的ip。

作者 east

bug清单, python 2月 15,2021

python3解决解决Please use the NLTK Downloader to obtain the resource

运行环境是windows，python3，运行python程序出现下面的错误：

Please use the NLTK Downloader to obtain the resource:

[31m>>> import nltk
>>> nltk.download(‘stopwords’)
[0m
Searched in:
– ‘C:\\Users\\99386/nltk_data’
– ‘C:\\nltk_data’
– ‘D:\\nltk_data’
– ‘E:\\nltk_data’
– ‘F:\\Program Files\\Python\\Python36\\nltk_data’
– ‘F:\\Program Files\\Python\\Python36\\lib\\nltk_data’
– ‘C:\\Users\\99386\\AppData\\Roaming\\nltk_data’

>>>python

>>> import nltk
>>> nltk.download(‘punkt’)

按照上面的命令会安装失败。要采用下面变通的方式：

接下来我们去https://github.com/nltk/nltk_data下载我们需要的数据。

只需下载正数第二个“packages”文件夹即可。把packages文件夹下的所有子文件夹拷贝至上面报错一个路径，例如F:\\Program Files\\Python\\Python36\\nltk_data 。请注意，不要把packages拷过去，而是packages的子文件夹。还有注意包含子文件夹，要把有些压缩包解压，例如把tokenizers下的punkt.zip解压（注意不要多了一层嵌套）。这时运行python程序可以看到原来的问题解决了。

作者 east

bug清单, shardingsphere 2月 13,2021

解决DATE_FORMAT(data_create_time, ‘%Y-%m-%d’)>= DATE_FORMAT(?, ‘%Y-%m-%d’)报错

使用 shardingsphere 4.1.0版本，代码如下：

 PreparedStatement ps = conn.prepareStatement("select * from test where DATE_FORMAT(create_date, '%Y-%m-%d')>= DATE_FORMAT(?, '%Y-%m-%d') limit ?,?");
    ps.setString(1,"2020-02-01");
    ps.setInt(2,1);
    ps.setInt(3,10)

报以下的错：

java.lang.ClassCastException: java.lang.String cannot be cast to java.lang.Integer
at org.apache.shardingsphere.sql.parser.binder.segment.select.pagination.PaginationContext.getValue(PaginationContext.java:57) ~[shardingsphere-sql-parser-binder-4.1.0.jar:4.1.0]
at org.apache.shardingsphere.sql.parser.binder.segment.select.pagination.PaginationContext.(PaginationContext.java:50) ~[shardingsphere-sql-parser-binder-4.1.0.jar:4.1.0]
at org.apache.shardingsphere.sql.parser.binder.segment.select.pagination.engine.LimitPaginationContextEngine.createPaginationContext(LimitPaginationContextEngine.java:38) ~[shardingsphere-sql-parser-binder-4.1.0.jar:4.1.0]
at org.apache.shardingsphere.sql.parser.binder.segment.select.pagination.engine.PaginationContextEngine.createPaginationContext(PaginationContextEngine.java:48) ~[shardingsphere-sql-parser-binder-4.1.0.jar:4.1.0]
at org.apache.shardingsphere.sql.parser.binder.statement.dml.SelectStatementContext.(SelectStatementContext.java:100) ~[shardingsphere-sql-parser-binder-4.1.0.jar:4.1.0]
at org.apache.shardingsphere.sql.parser.binder.SQLStatementContextFactory.getDMLStatementContext(SQLStatementContextFactory.java:103) ~[shardingsphere-sql-parser-binder-4.1.0.jar:4.1.0]
at org.apache.shardingsphere.sql.parser.binder.SQLStatementContextFactory.newInstance(SQLStatementContextFactory.java:87) ~[shardingsphere-sql-parser-binder-4.1.0.jar:4.1.0]
at org.apache.shardingsphere.underlying.route.DataNodeRouter.createRouteContext(DataNodeRouter.java:99) ~[shardingsphere-route-4.1.0.jar:4.1.0]
at org.apache.shardingsphere.underlying.route.DataNodeRouter.executeRoute(DataNodeRouter.java:89) ~[shardingsphere-route-4.1.0.jar:4.1.0]
at org.apache.shardingsphere.underlying.route.DataNodeRouter.route(DataNodeRouter.java:76) ~[shardingsphere-route-4.1.0.jar:4.1.0]

经过测试，传递的参数如果为string类型就报下面的错误，如果把类型修正为Date类型则不会报如下错误。配置：

spring:
  shardingsphere:
    encrypt:
      encryptors:
        aesEncryptor:
          type: aes
          props:
            aes.key.value: 123456
      tables:
        linkman:
          columns:
            mobile:
              plainColumn: mobile
              cipherColumn: mobile_encrypted
              encryptor: aesEncryptor
    datasource:
      names: ds
      ds:
        type: com.zaxxer.hikari.HikariDataSource
        driver-class-name: com.mysql.cj.jdbc.Driver
        jdbc-url: jdbc:mysql://localhost:3306/shardingsphere_test?useUnicode=true&characterEncoding=utf8&allowMultiQueries=true&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8&autoReconnect=true
        username: root
        password: root
        connection-timeout: 30000
        maximum-pool-size: 20
        minimum-idle: 5
        auto-commit: true
        idle-timeout: 600000
        pool-name: crm-hikari
        max-lifetime: 1800000
        connection-test-query: SELECT 1
    props:
      query.with.cipher.column: false

<select id="selectByTime" resultMap="BaseResultMap"  parameterType="string">
        SELECT  <include refid="Base_Column_List"/>
         FROM linkman
         WHERE DATE_FORMAT(create_time, '%Y-%m-%d') =  DATE_FORMAT(#{createTime},'%Y-%m-%d')
    </select>

    List<Linkman> selectByTime(@Param("createTime") String createTime);
//如果把createTime的类型修改为Date类型则不会报错.

作者 east

python 2月 13,2021

ES无法重启问题分析

题背景：

对ES集群进行了重启，集群重启几分钟后，部分实例开始逐渐下线，导致集群不可恢复。

集群规模：

普通模式，3EsMaster，40EsNode，每实例均为31GB内存。

数据量：

1000多index,38365个shard，其中主分片28695个，数据量100T。

日志分析：

ES集群重启，恢复几分钟后，出现大量ping Master节点超时的错误，然后ES节点实例开始逐渐下线，导致集群恢复不了。

原因分析：

集群分片数过多，并发恢复过程中，同时业务没有停止，导致EsMaster处理压力过大，number_of_pending_tasks(挂起的任务)逐渐增加，到达4W多，大量的任务阻塞。

此时_cluster/health命令已无法正常返回结果，导致大量ES实例处于恢复中状态，连续3次检查超时后，Manger将会重启实例。就会导致挂起的任务越来越多，集群不可恢复。

处理步骤：

重启的过程中，发现大量ES实例执行_cluster/health命令超时错误，但其实数据仍在缓慢恢复中。于是注释_cluser/health检查脚本，防止实例多次失败后，被manger重启。
再次重启后，发现大量分片处于未分配状态，执行_cluster/allocation/explain查看分片未被分配的原因，发现shard恢复时的cluster.routing.allocation.node_current_incoming达到了最大值。鉴于集群主分片数太多，于是调大恢复参数至：cluster.routing.allocation.node_initial_primaries_recoveries 200 cluster.routing.allocation.node_concurrent_recoveries 100 cluster.routing.allocation.cluster_concurrent_rebalance 100 同时因为有业务数据写入，将分片分配设置为none:cluster.routing.allocation.enable none
再次重启后，集群开始恢复，查看_cat/thred_pool/，generic 线程池（分片恢复会用到该线程池）已经到达128，查看界面CPU使用率也在70-80，查看日志，分片正在恢复中。
10分钟左右，集群恢复到80%左右，开始恢复缓慢，初始化分片（initializing_shards）有2000多个，这些分片初始化的过程耗时接近2小时原因分析：因为在多次重启的过程中，业务侧并没有停止，由于有些primary新写入了数据，在数据的recovery过程中，需要从主副本之间拷贝数据，或者利用translog恢复数据。直达primary-replica完全in-sync后，才会完成初始化。这个过程取决于shard的大小和新写入量的大小（初始化的分片普遍数据量较大）。
1. 最后有一个分片无法分配，查看原因，该分片无法从translog in-sync（同步），查看该索引settings，sync_interval设置为360s，设置同步刷新时间过多，会有一定几率发生数据丢失（客户有原数据备份）。
2. 集群恢复后，还原集群参数配置和健康检查脚本。

问题根因：

集群的分片数过多，按一个实例管理600shard为标准，该集群分片数过度超标的。合理设置索引分片，定期对历史索引进行处理（关闭或删除不需要的索引）。
目前的健康检查机制需要优化，使用_cluster/health去判断各实例的健康是否合理，包括检查周期等。在EsMaster压力过大的情况下，_cluster/health可能会造成误判。

作者 east

bug清单, Elasticsearch 2月 12,2021

ES实例磁盘空间不足，导致索引read-only

题现象：

数据导入失败错误日志：

retrying failed action with response code: 403 ({“type”=>”cluster_block_exception”, “reason”=>”blocked by: [FORBIDDEN/12/index read-only / allow delete (api)]

问题原因：

磁盘空间不足，磁盘使用率大于95%，索引被强制设置为只读，导致数据无写入。

解决办法：

1. 下线磁盘空间不足的ES实例。

curl -XPUT --negotiate -k -u : "https://127.0.0.1:24100/_cluster/settings" -H 'Content-Type: application/json'  -d '{
"transient" : {
"cluster.routing.allocation.exclude._name" : "EsNode1@192.168.198.6, EsNode2@192.168.198.6 "
}
}'

2. 等待数据迁移完成。执行_cluster/healht命令查询relocating_shards参数的值，直到变为0，说明迁移完毕。

curl -XGET --negotiate -k -u : "https://127.0.0.1:24100/_cluster/health?pretty"

3. 修改索引只读字段属性为null，放开写入。

curl -XPUT --negotiate -k -u : "https://127.0.0.1:24100/*/_settings" -H 'Content-Type: application/json' -d '{"index.blocks.read_only_allow_delete": null}'

4. 继续写入数据。

作者 east

bug清单, Elasticsearch 2月 12,2021

es bulk入库数据丢失分析思路

bulk入库数据丢失

【问题现象】

用户反馈数据，会生成报文，每天丢20%的报文，ES中存储的是2种日志，通过抽了2个业务日志做比对，反馈说有一些流水的号没有打出来，怀疑有丢失数据的情况

【分析过程】

建议业务单独入库疑似丢掉的数据，但是数据已经丢失，没法复现。

后续了解到业务是通过bulk批量方式入库的，但是bulk方式入库会有个现象：批量入库后会返回整体的成功信息，如果此批次里面有错误数据的话是不会报错的，同时也不会入库错误数据，带来的现象就是数据丢失。

随即本地模拟了bulk方式入库，如下入库3条数据，在其中构造出一条错误数据，可以看到整体是反馈成功的。

bulk入库数据丢失

后续通过查询索引只能查询出2条数据,如下：

bulk入库数据丢失

【处理建议】

1．从入库数据源头提前了解到数据信息，防止错误数据入库。

2．可以对代码二次开发进行每条入库数据解析，能及时观察到错误数据。

3．单独入库每条数据，会对性能有较大影响。

作者 east

技术架构 2月 12,2021

直播间机器人设计方案

设计目标

1：模拟一定人数在房间，保证房间的人气

2：不同时段房间内房间机器人的人数要有波动，防止人数不真实

3：支持房间内做活动，需要临时增加大批量的用户

4：机器人不能出现大批量进入房间，大批量退出房间的情形

设计流程

1：增加robot进程进行通知机器人进入房间退出房间的处理

2：设计给机器人分配特定的号段区间（例如：850000000-860000000）

2：robot进程读取配置文件，获取房间信息，以及房间的机器人配置信息流程

4：如有讲师在麦上，robot服务定时判断真实人数和在线人数的比例公式,判断是否需要增加或减少机器人，如果需要，则发起信令请求，消息流为：

robot->usermgrsvr->gateway->roomsvr

5:房间服务器获取到加入房间请求后，判断当前的最大机器人ID,然后从最大的机器人ID

往上累加，获取最新的机器人ID。（如果达到最高的配置860000000，则从最低850000000开始循环）。然后根据新增加机器人ID，做加入房间的处理.

6:房间服务器获取到退出房间的请求后，判断当前的最小机器人ID，然后删除指定的机器人数量

7: 如果没有讲师在麦上，robot服务逐步删除房间内的机器人，发起信令请求，消息流为：

robot->usermgrsvr->gateway->roomsvr，房间服务器处理为第6部。

配置文件

robot配置文件robot.conf如下信息：

[roomlist] #房间列表配置

Roomid=60001,60002,60003

[onlineparam]

onlinenum=20 #机器人基础在线人数

onlinerate=1-200:30;201-500:20;501-2000:10;2001-1000000000:5 #机器人数比率

假设M为真实的在线总人数

当1<=M<=200时

Totalonline = onlinenum + M * 30

当200<M<=500时

Totalonline = onlinenum + 200 * 30 + (M-200) * 20

当500<M<=2000时

Totalonline = onlinenum + 200 * 30 + 300 * 20 + (M-500) * 10

当2000<M<=1000000000时

Totalonline = onlinenum + 200 * 30 + 300 * 20 + 500 * 10 + (M-500) *5

作者 east