pyflink 写ES并发和串行-Toy模板网

这篇具有很好参考价值的文章主要介绍了pyflink 写ES并发和串行。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

写ES并发执行:
# -*- coding: utf-8 -*-
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.functions import MapFunction, RuntimeContext, KeyedProcessFunction
from abc import ABC, abstractmethod
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.functions import MapFunction, RuntimeContext, KeyedProcessFunction
from pyflink.datastream.state import MapStateDescriptor
from pyflink.datastream.connectors.kafka import FlinkKafkaConsumer
from pyflink.common.typeinfo import Types, TypeInformation
from pyflink.datastream.connectors.elasticsearch import Elasticsearch7SinkBuilder, ElasticsearchEmitter, FlushBackoffType
from pyflink.datastream.connectors import DeliveryGuarantee
from pyflink.common.serialization import SimpleStringSchema
import json
import re
from datetime import datetime
from elasticsearch import Elasticsearch

import re
import redis

# 创建 StreamExecutionEnvironment 对象
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)

# 读取文件，创建 DataStream 对象
data_stream = env.read_text_file('/root/pyflink/elink_test.txt')
# 对每行数据添加字符串 'aaaa'
class LogEvent:
bus_seq = None
line_number = None
event_received_time = None
app_name = None
source_module_name = None
source = None
filename = None
message = None
serial_id = None
thread_id = None
class MyMapFunction(MapFunction):
def open(self, runtime_context: RuntimeContext):
pool = redis.ConnectionPool(host='127.0.0.1',port=6379,max_connections=50)
self.r = redis.Redis(connection_pool=pool)
def close(self):
self.r.close()

def map(self,line):
process_id='';
bus_seq=''
if not line.startswith("ES"):
return
if '<Serial>' in line:
pat=re.compile(r"<Serial>(\d+)</Serial>")
bus_seq=pat.findall(line)
process_id=line.split()[1]
self.r.set(process_id,bus_seq[0])
process_id=line.split()[1]
if not len(process_id)==6 :
process_id=line.split()[2]

bus_seq=self.r.get(process_id)
if not bus_seq:
return
#self.r.delete(process_id)
log_event = LogEvent()
LogEvent.bus_seq=bus_seq.decode('UTF-8')
LogEvent.message=line
#return (log_event.bus_seq.decode('UTF-8'),log_event.message)
return LogEvent

class EsSink(MapFunction):
def open(self, runtime_context: RuntimeContext):
self.es = Elasticsearch("http://127.0.0.1:9200")

def close(self):
pass

def map(self,LogEvent):
try:
data = {
"@timestamp": datetime.now().strftime( "%Y-%m-%dT%H:%M:%S.000+0800" ),
"content" : LogEvent.message,
"bus_seq" : LogEvent.bus_seq
}
except:
return
self.es.index( index="flink_test", document=data )

new_stream = data_stream.map(MyMapFunction()).map(EsSink()).set_parallelism(3)

# 输出到控制台
#new_stream.print()

# 执行任务
env.execute('Add "bus_seq" to each line')

88行:
self.es.index( index="flink_test", document=data )

[root@master pyflink]# python process_log.py
process_log.py:88: ElasticsearchWarning: Elasticsearch built-in security features are not enabled. Without authentication, your cluster could be accessible to anyone. See https://www.elastic.co/guide/en/elasticsearch/reference/7.17/security-minimal-setup.html to enable security.
self.es.index( index="flink_test", document=data )
process_log.py:88: ElasticsearchWarning: Elasticsearch built-in security features are not enabled. Without authentication, your cluster could be accessible to anyone. See https://www.elastic.co/guide/en/elasticsearch/reference/7.17/security-minimal-setup.html to enable security.
self.es.index( index="flink_test", document=data )
process_log.py:88: ElasticsearchWarning: Elasticsearch built-in security features are not enabled. Without authentication, your cluster could be accessible to anyone. See https://www.elastic.co/guide/en/elasticsearch/reference/7.17/security-minimal-setup.html to enable security.
self.es.index( index="flink_test", document=data


写ES串行执行:

[root@master pyflink]# python process_log.py
process_log.py:88: ElasticsearchWarning: Elasticsearch built-in security features are not enabled. Without authentication, your cluster could be accessible to anyone. See https://www.elastic.co/guide/en/elasticsearch/reference/7.17/security-minimal-setup.html to enable security.
self.es.index( index="flink_test", document=data )

  文章来源地址https://www.toymoban.com/news/detail-472378.html

到了这里，关于pyflink 写ES并发和串行的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！