ELK下载
地址：Past Releases of Elastic Stack Software | Elastic
在Products和version处分别选择需要下载的产品和版本，E（elasticsearch）L（logstash）K（kibana）三者版本必须相同
将下载好的elk分别解压到相同路径下
本文中elasticsearch=E=ES=es；L=logstash；K=kibana
配置
E配置
一般情况下使用默认配置即可，下面对我的部分配置进行简单描述
network.host:0.0.0.0 //0.0.0.0表示可以连任意地址
http.port:9200 //表示es的访问端口
上图默认配置文件中没有 这些是 Elasticsearch 的安全性设置，并且必须在您的 elasticsearch.yml 配置文件中设置。这是它们的具体含义：
xpack.security.enabled: true
#启用 Elasticsearch 的内置安全功能。
xpack.license.self_generated.type: basic
#安装基本许可证以启用 Ealsticsearch 的基本内置功能。
#这种许可证是免费的，允许使用一些基本的安全功能，例如用户身份验证。
xpack.security.transport.ssl.enabled: true
#启用 Elasticsearch 节点之间的 SSL 加密通道，从而增强数据传输的安全性。
注意，要使这些设置生效，必须在配置文件中启用 X-Pack，并设置相关的用户名、密码等属性。此外，确保您的 Elasticsearch 版本支持这些安全设置。
要确定是否在 Elasticsearch 配置文件中启用了 X-Pack，并设置相关的用户名、密码等属性，请按照以下步骤：
1. 打开 Elasticsearch 的安装目录并找到 config 目录。
2. 打开 elasticsearch.yml 文件。
3. 搜索 `xpack.security.enabled` 属性并确保其值为 `true`。如果没有此属性，则意味着未启用 X-Pack 安全功能。
4. 搜索 `xpack.security.authc.*` 属性。这些属性定义了用于身份验证和授权的用户名和密码等详细信息。如果没有这些属性，则意味着未设置相关的认证信息。
5. 根据您的需求设置适当的身份验证和授权方案，然后保存 elasticsearch.yml 文件并重新启动 Elasticsearch。确保在重新启动 Elasticsearch 之前，所有配置的用户名和密码都是安全的，并将其作为文件或环境变量等方式保存。
上述安全性设置非必要可不开，开启后若不配置用户名密码等kibana无法访问
K配置
默认配置即可，若要上传的数据量过大，可修改如下配置，此处我将默认1GB改为了50GB
server.maxPayload: 536870912
#可根据需求自行修改大小下面我给出一些其他配置说明
server.host: 您可以通过该配置项指定 Kibana 服务的监听地址，例如：
server.host: "0.0.0.0" # 监听所有网络接口
server.host: "localhost" # 仅监听本机
server.host: "192.168.0.1" # 监听特定网络接口server.port: 您可以通过该配置项指定 Kibana 服务的监听端口，默认为 5601。
elasticsearch.url: 您需要指定 Elasticsearch 服务的访问路径和端口，例如：
elasticsearch.url: "http://localhost:9200"
kibana.index: 您可以指定 Kibana 的索引名称，默认为 .kibana。如果您想将索引存储在远程 Elasticsearch 集群中，可以使用以下配置项：
kibana.index: ".mykibanaindex"
elasticsearch.hosts: ["http://localhost:9200"]
logging: 您可以指定 Kibana 日志级别和输出方式，例如：
logging.dest: stdout # 输出到控制台
logging.quiet: true # 禁止输出所有日志信息
logging.verbose: true # 输出详细的日志信息
设置好配置文件之后，您可以使用以下命令启动 Kibana：
bin/kibana
请注意，Kibana 的具体配置可能因版本而异，这里提供的只是一些常见的配置。如果您需要更多的配置选项，建议查阅 Kibana 的官方文档。
启动
先启动E，在es安装目录下找到bin目录下的elasticsearch.bat，双击启动
接着启动K，在kibana安装目录下找到bin目录下的kibana.bat，双击启动
启动后命令行界面不要关闭，若要停止Ctrl+C即可，或者直接叉掉。可以选择将其设置为Windows服务，更方便一点，不设置也行，无所屌谓
启动后在浏览器输入地址：
http://localhost:5601
#5601是kibana默认端口，若自己设置了端口，用自己配置打开即可
这是kibana数据可视化界面
如果阁下英文柔弱，那么我来教阁下如何汉化以至于应对此问题
在kibana配置文件kibana.yml中添加如下配置，重启生效
i18n.locale: "zh-CN"
#zh-CN是什么意思呢？即简体-中文将CSV文件通过kibana创建索引导入数据
在kibana可视化界面点击设置→索引模式→上传文件创建索引即可
默认最大限制上传100MB文件，若文件大于100MB，可在设置→高级设置→最大文件上传大小设置为1GB
若1GB还是不够，我来教阁下如何应对
修改es配置文件elasticsearch.yml，添加如下配置
http.max_content_length: 2gb
#这里设置了2gb，阁下可自行调整其他大小那么问题来了，如果配置文件中最大文件上传大小限制修改过大后，会导致JVM虚拟机内存不足，以至于启动es的时候闪退崩掉，此时需要修改es安装目录下的config目录下的jvm.options文件
将-Xms和-Xmx设置为自己需要的大小，-Xms为最小内存，-Xmx为最大内存
设置完成后重启即可
如果上面的方法行不通，别急，还有办法，且看下文分解
既然大的csv文件上传不了，那咱就换个方式，规矩是死的但人是活的，正所谓事在人为
将大的csv文件拆分成若干个小的csv文件进行上传，然后再将这些上传的小csv数据合并
首先打开powershell，输入以下代码
$filePath = "F:\bigdata\naraka.csv"
#F:\bigdata\naraka.csv是你要拆分的csv文件的所在路径
$outputPath = "F:\bigdata\splitted"
#F:\bigdata\splitted是拆分后小文件们的存放路径
#注意看，这个文件夹叫splitted，需要提前创建，当然你也可以取自己喜欢的名字
$filePrefix = "naraka_"
#naraka_是拆分后的小文件们的名字，例如：naraka_1、naraka_2
#你也可以取自己喜欢的名字
$fileSize = 100MB
#这是拆分成小文件的大小，你可以自行修改所拆分的文件大小
#这里设置的是每个小csv文件为100MB
$fileCount = 1
#这里设置的是拆分后的小文件的命名编号从1开始
#下面是拆分逻辑，直接用，没啥可修改的地方
$stream = [System.IO.StreamReader]::new($filePath)
try {
$buffer = [char[]]::new(1MB)
while (!$stream.EndOfStream) {
$splittedFile = Join-Path $outputPath "$filePrefix$fileCount.csv"
$writer = [System.IO.StreamWriter]::new($splittedFile, $false, [System.Text.Encoding]::UTF8)
$remaining = $fileSize
while ($remaining) {
$readSize = [Math]::Min($buffer.Length, $remaining)
$readCount = $stream.ReadBlock($buffer, 0, $readSize)
if (!$readCount) {
break
}
$writer.Write($buffer, 0, $readCount)
$remaining -= $readCount
}
$writer.Flush()
$writer.Dispose()
$fileCount++
}
} finally {
$stream.Dispose()
}执行完毕以后到对应的目录下就可以看到以及拆分成了若干个小文件
拆分成小文件以后就可以愉快的导入es并创建索引啦
当然，如果不想根据文件大小来拆分文件或者由于数据问题想要根据文件内数据的行数来拆分的话，可以使用如下power shell命令
需要先修改一下原来的拆分点查找逻辑，改为通过分隔符进行查找。然后可以在处理每行数据时，记录当前行号并进行计数。达到800000行时，即可开始拆分新的文件，修改后的代码如下：
$filePath = "F:\bigdata\naraka.csv"
$outputPath = "F:\bigdata\splitted"
$filePrefix = "naraka_"
$lineCountToSplit = 800000
#一个文件数据达到800000行时，拆分为一个文件
#第800001行则成为拆分后第二个csv文件的第一行，以此类推
$fileCount = 1
$stream = [System.IO.StreamReader]::new($filePath)
try {
$buffer = [char[]]::new(1MB)
while (!$stream.EndOfStream) {
$splittedFile = Join-Path $outputPath "$filePrefix$fileCount.csv"
$writer = [System.IO.StreamWriter]::new($splittedFile, $false, [System.Text.Encoding]::UTF8)
$remaining = $lineCountToSplit
$currentLine = 1
while ($remaining -ge 0 -and !$stream.EndOfStream) {
$readSize = [Math]::Min($buffer.Length, $remaining)
$readCount = $stream.ReadBlock($buffer, 0, $readSize)
if (!$readCount) {
break
}
$text = [System.String]::new($buffer, 0, $readCount)
$lines = $text.Split([System.Environment]::NewLine, [System.StringSplitOptions]::RemoveEmptyEntries)
foreach ($line in $lines) {
if ($currentLine -ne 1) {
$remaining--
}
$writer.WriteLine($line)
if ($remaining -eq 0) {
$writer.Flush()
$writer.Dispose()
$fileCount++
$splittedFile = Join-Path $outputPath "$filePrefix$fileCount.csv"
$writer = [System.IO.StreamWriter]::new($splittedFile, $false, [System.Text.Encoding]::UTF8)
$remaining = $lineCountToSplit
}
$currentLine++
}
$remaining -= $readCount
}
$writer.Flush()
$writer.Dispose()
$fileCount++
}
} finally {
$stream.Dispose()
}这个修改会在每800000行数据处理完毕后，进行拆分新的文件，并继续处理剩余数据。在处理过程中，会记录当前行号并进行计数，以免在处理行数不足800000时，出现拆分位置计算异常的问题
既然上面代码是根据文件的行数来进行拆分，那可不可以通过列来拆分文件呢？答案是那当然啦
在上述代码的基础上，根据需求，需要在处理时识别CF列的位置，并从该列之后开始进行拆分文件，下面是实现代码：
$filePath = "F:\bigdata\naraka.csv"
#所要被拆分的文件
$outputPath = "F:\bigdata\splitted"
#拆分后的文件的存储路径
$filePrefix = "naraka_"
#拆分后的文件的命名格式
$fileSize = 200MB
#每个拆分后的文件的大小
$fileCount = 1
#拆分后的文件的命名编号从1开始
$splitColumn = "CF"
#从CF列拆分
$stream = [System.IO.StreamReader]::new($filePath)
try {
$buffer = [char[]]::new(1MB)
while (!$stream.EndOfStream) {
$splittedFile = Join-Path $outputPath "$filePrefix$fileCount.csv"
$writer = [System.IO.StreamWriter]::new($splittedFile, $false, [System.Text.Encoding]::UTF8)
$remaining = $fileSize
$splitPoint = $stream.ReadLine().IndexOf($splitColumn)
while ($remaining) {
$readSize = [Math]::Min($buffer.Length, $remaining)
$readCount = $stream.ReadBlock($buffer, 0, $readSize)
if (!$readCount) {
break
}
$text = [System.String]::new($buffer, 0, $readCount)
$lines = $text.Split([System.Environment]::NewLine, [System.StringSplitOptions]::RemoveEmptyEntries)[1..-1]
foreach ($line in $lines) {
$splitIndex = $line.IndexOf(',', $splitPoint)
$isNewFile = $splitIndex - $splitPoint - 1 + $writer.BaseStream.Position > $fileSize
if ($isNewFile) {
$remaining = 0
break
}
$writer.WriteLine($line)
}
$remaining -= $readCount
}
$writer.Flush()
$writer.Dispose()
$fileCount++
}
} finally {
$stream.Dispose()
}在修改后的代码中，首先定义了一个$splitColumn变量，用于指定拆分点所在的列。然后，在每次读取和写入文件时，我们将取出剩余缓存中的内容，按行进行拆分，并逐行处理文本内容。在处理时，我们使用$splitPoint变量获取拆分点位置，并将每行数据按该位置进行拆分处理。如果当前行数据跨越了拆分点的位置，则将其写入下一个文件中，否则仍然写入当前文件
如果你的文件的数据内容比较特殊，免不了会出现异常，这个时候问题又来了，因为数据内容的原因，拆分点处如果找不到这个列时应该怎么办？出现的异常通常是因为字符串中没有找到指定字符而引发的，下面继续教阁下如何应对：
可以先使用$splitColumn找到指定列的位置，确保该列存在并且位置正确。如果该列存在，那么在执行$line.IndexOf时，需要传入正确的startIndex参数，以限定寻找字符的位置范围，下面是修改后的代码，增加了一些判断和调整：
$filePath = "F:\bigdata\naraka.csv"
$outputPath = "F:\bigdata\splitted"
$filePrefix = "naraka_"
$fileSize = 200MB
$fileCount = 1
$splitColumn = "CF"
$stream = [System.IO.StreamReader]::new($filePath)
try {
$buffer = [char[]]::new(1MB)
while (!$stream.EndOfStream) {
$splittedFile = Join-Path $outputPath "$filePrefix$fileCount.csv"
$writer = [System.IO.StreamWriter]::new($splittedFile, $false, [System.Text.Encoding]::UTF8)
$remaining = $fileSize
$splitPoint = $null
$header = $stream.ReadLine()
$splitPoint = $header.IndexOf($splitColumn)
while ($remaining) {
$readSize = [Math]::Min($buffer.Length, $remaining)
$readCount = $stream.ReadBlock($buffer, 0, $readSize)
if (!$readCount) {
break
}
$text = [System.String]::new($buffer, 0, $readCount)
$lines = $text.Split([System.Environment]::NewLine, [System.StringSplitOptions]::RemoveEmptyEntries)[1..-1]
foreach ($line in $lines) {
$splitIndex = $line.IndexOf(',', $splitPoint+1)
if ($splitIndex - $splitPoint > 0) {
$isNewFile = $splitIndex - $splitPoint - 1 + $writer.BaseStream.Position > $fileSize
}
else {
$isNewFile = $line.Length - $splitPoint + $writer.BaseStream.Position > $fileSize
}
if ($isNewFile) {
$remaining = 0
break
}
$writer.WriteLine($line)
}
$remaining -= $readCount
}
$writer.Flush()
$writer.Dispose()
$fileCount++
}
} finally {
$stream.Dispose()
}在修改后的代码中，增加了以下内容：
在每次处理新的拆分文件前，先读取文件的第一行内容，以获取$splitPoint拆分点所在位置。在找不到拆分点时，可以通过正确检查该列信息，来发现问题所在。
在$line.IndexOf中，将startIndex参数加一，以跳过实际拆分点的位置。
在处理过程中，如果找不到拆分点，需要重新计算$isNewFile以判断是否需要切换到下一个拆分文件。
通过上述种种方法，导入小文件以后可以在kibana的 设置→索引管理 里面通过“强制合并索引”功能来将它们合并，当然不建议采取这种方法，因为这个必须先保证你的每个小文件创建的索引字段一致，当然既然不建议采取这种方法，接下来总归是会提供别的方法
那就是可以使用 Elasticsearch 的 Bulk API，将这些小文件合并成单个索引。这种方法可以保持数据的完整性和一致性。注意，数据块的大小仍然需要考虑内存大小、网络带宽以及索引性能等方面的因素。
由于本人所要处理的数据量过于巨大，拆分成若干个小文件处理的话，一方面是比较麻烦，再一方面数据内容比较特殊，因此采用 数据库→es 的方式，即使用logstash进行。
首先，下载logstash，解压到es和kibana相同路径下，上文有所述。
L配置
使用默认配置即可启动，在logstash安装目录的bin下点击logstash.bat即可启动，注意需先启动es，再启动logstash
使用默认配置启动时可能会出现如下问题
出现这个警告是因为logstash自带了JDK，而这里使用了系统本身配置的JDK，如果可以正常启动则忽略即可
由于需要使用logstash将mysql数据库某个表的数据上传到es并创建索引，因此使用默认配置启动肯定是不行滴，话不多说，开教：
在logstash安装目录的config文件夹下新建logstash.conf文件，原来肯定是没有这个文件的，只有logstash.yml和一个logstash-example.conf文件。
新建好logstash.conf后，打开它进行相关配置：
input {
jdbc {
jdbc_driver_library => "E:\mysql\mysql-connector-java-8.0.24.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://0.0.0.0:0000/naraka"
jdbc_user => "root"
jdbc_password => "******"
schedule => "*/1 * * * *"
statement => "SELECT * FROM name"
use_column_value => true
tracking_column => "id"
}
}
output {
elasticsearch {
hosts => ["localhost:9200"]
index => "naraka"
document_id => "%{id}"
}
}需要替换以下内容：
- jdbc_driver_library: MySQL connector jar包的路径
如果找不到这个包也没关系，自行去mysql官网下载即可，路径随意，填写正确即可
- jdbc_driver_class指定驱动程序的类型
- jdbc_connection_string: MySQL连接字符串，确保连接到正确的数据库
- jdbc_user和jdbc_password: MySQL数据库的账号和密码
- statement: 要执行的查询语句，这里查询name表的所有数据
- index: Elasticsearch中的索引名
- document_id: 每条数据在Elasticsearch中的唯一标识，这里使用name表中的id字段值作为标识
启动
配置完成后，打开命令行，进入Logstash的bin目录，执行以下命令启动Logstash：
logstash -f "path/to/logstash.conf"
#"path/to/logstash.conf"是所新建的logstash.conf所在的路径，自行修改即可确保替换logstash.conf为实际的Logstash配置文件名
查看Logstash日志，确认数据已经上传到Elasticsearch中。
可以使用以下命令查询Elasticsearch中的数据：
curl -XGET localhost:9200/naraka/_search替换naraka为实际的索引名
欸嘿，既然看到这里，说明一定是出了一些问题
注意看，上述配置只适合数据量小的情况，如果数据量大，则会报出以下问题：
[ERROR][logstash.inputs.jdbc ][main][azq12wsxcde34rfvbgt56yhnmju78iklo90p] Java::JavaSql::SQLException: Java heap space: SELECT * FROM name
这个错误提示来自于 Logstash，它在执行一个查询语句时遇到了 Java 堆空间不足的问题，无法继续执行。查询语句是在数据库中选取 “name” 表中的所有数据（SELECT * FROM name）
出现Java堆空间不足的原因是Logstash在进行数据处理时，需要使用的内存超过了JVM分配给Logstash的堆空间大小。这可能是由于Logstash正在处理大量数据或者配置不当导致的。
要解决这个问题，首先可以尝试增加Logstash的JVM堆空间大小。可以通过编辑启动脚本，例如logstash.bat或者logstash.sh文件，在其中增加JVM参数，如-Xmx和-Xms来调整堆空间大小。另外，如果是因为处理大量数据导致的问题，可以考虑使用更高级别的硬件配置或者采用分批次处理等策略来减少内存压力。
当然，有无别的方法可以在不修改JVM的前提下解决这个问题呢？本人更推荐下面这种方法：
由于上述问题是使用配置文件跑全量时产生的，那么我们只需要在配置文件中添加一些分批的指令就好啦，如下：
input {
stdin {
}
jdbc {
jdbc_driver_library => "E:\mysql\mysql-connector-java-8.0.24.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://0.0.0.0:0000/naraka"
jdbc_user => "root"
jdbc_password => "******"
jdbc_paging_enabled => "true"
jdbc_page_size => "50000"
schedule => "*/1 * * * *"
statement => "SELECT * FROM name"
use_column_value => true
tracking_column => "id"
}
}
output {
elasticsearch {
hosts => ["localhost:9200"]
index => "naraka"
document_id => "%{id}"
}
stdout {
codec => json_lines
}
}需要替换以下内容：
- jdbc_driver_library: MySQL connector jar包的路径
如果找不到这个包也没关系，自行去mysql官网下载即可，路径随意，填写正确即可
- jdbc_driver_class指定驱动程序的类型
- jdbc_connection_string: MySQL连接字符串，确保连接到正确的数据库
- jdbc_user和jdbc_password: MySQL数据库的账号和密码
- statement: 要执行的查询语句，这里查询name表的所有数据
- index: Elasticsearch中的索引名
- document_id: 每条数据在Elasticsearch中的唯一标识，这里使用name表中的id字段值作为标识
在上述代码中，增加了
jdbc_paging_enabled和jdbc_page_size，指定了Logstash如何处理查询到的结果集
stdin {}表示从标准输入读取数据的插件
stdout {}表示将数据输出到标准输出中并以 JSON 格式展示，可用于测试和调试
这个Logstash配置文件，它负责将从MySQL数据库中查询到的数据推送到Elasticsearch索引中，输入部分定义了两个输入，一个从标准输入（stdin）中读取数据，另一个是使用JDBC从MySQL数据库中读取数据；输出部分定义了两个输出，一个将结果推送到Elasticsearch索引中，其中hosts指定Elasticsearch服务器的位置，index指定索引名称，document_id指定每个文档的id。另一个是将结果输出到标准输出（stdout）文章来源：https://www.toymoban.com/news/detail-613483.html
使用如上配置就可以处理数据量比较大的数据啦，静静等它跑完，To be continued......文章来源地址https://www.toymoban.com/news/detail-613483.html
到了这里，关于ELK（Elasticsearch、Kibana、Logstash）以及向ES导入mysql数据库数据或CSV文件数据，创建索引和可视化数据的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
