西安阿里云网站建设,wordpress通知评论者,装饰公司手机网站建设,国家能源局网站线路建设1.Maxwell的配置 其实就是配置两端的配置信息,都要能连接上,然后才能去传输数据 config.properties
#Maxwell数据发送目的地#xff0c;可选配置有stdout|file|kafka|kinesis|pubsub|sqs|rabbitmq|redis
producerkafka
# 目标Kafka集群地址
kafka.bootstrap.servershadoop102…1.Maxwell的配置 其实就是配置两端的配置信息,都要能连接上,然后才能去传输数据 config.properties
#Maxwell数据发送目的地可选配置有stdout|file|kafka|kinesis|pubsub|sqs|rabbitmq|redis
producerkafka
# 目标Kafka集群地址
kafka.bootstrap.servershadoop102:9092,hadoop103:9092,hadoop104:9092
#目标Kafka topic可静态配置例如:maxwell也可动态配置例如%{database}_%{table}
kafka_topictopic_db# MySQL相关配置
hosthadoop102
usermaxwell
passwordmaxwell
jdbc_optionsuseSSLfalseserverTimezoneAsia/ShanghaiallowPublicKeyRetrievaltrue# 过滤gmall中的z_log表数据该表是日志数据的备份无须采集
filterexclude:gmall.z_log
# 指定数据按照主键分组进入Kafka不同分区避免数据倾斜
producer_partition_byprimary_key2.群起脚本
#!/bin/bashMAXWELL_HOME/opt/module/maxwell-1.29.2status_maxwell(){resultps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | wc -lreturn $result
}start_maxwell(){status_maxwellif [[ $? -lt 1 ]]; thenecho 启动Maxwell$MAXWELL_HOME/bin/maxwell --config $MAXWELL_HOME/config.properties --daemonelseecho Maxwell正在运行fi
}stop_maxwell(){status_maxwellif [[ $? -gt 0 ]]; thenecho 停止Maxwellps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | awk {print $2} | xargs kill -9elseecho Maxwell未在运行fi
}case $1 instart )start_maxwell;;stop )stop_maxwell;;restart )stop_maxwellstart_maxwell;;
esacKafka消费者开始等待消费
bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic topic_db4.全量同步
/opt/module/maxwell/bin/maxwell-bootstrap --database gmall --table user_info --config /opt/module/maxwell/config.properties每条数据都有时间戳,这个时间戳是同步的时间
5.为什么Kafka后面需要拦截器? 零点漂移,当为一天末点时,时间戳还是末点,但是到第二个Flume时,时间戳更新,这时候直接到第二天,导致数据与时间不一致.
