当前位置: 首页 > news >正文

Flume采集Kafka并把数据sink到OSS

安装环境

  1. Java环境, 略 (Flume依赖Java)
  2. Flume下载, 略
  3. Scala环境, 略 (Kafka依赖Scala)
  4. Kafak下载, 略
  5. Hadoop下载, 略 (不需要启动, 写OSS依赖)

配置Hadoop

下载JindoSDK(连接OSS依赖), 下载地址Github
解压后配置环境变量

export JINDOSDK_HOME=/usr/lib/jindosdk-x.x.x
export HADOOP_CLASSPATH=$HADOOP_CLASSPATH:${JINDOSDK_HOME}/lib/*

修改Hadoop配置, core-site.xml

<property><name>fs.oss.credentials.provider</name><value>com.aliyun.jindodata.oss.auth.SimpleCredentialsProvider</value></property><property><name>fs.oss.accessKeyId</name><value>xxxx</value></property><property><name>fs.oss.accessKeySecret</name><value>xxxx</value></property><property><name>fs.oss.endpoint</name><value>xxxxx</value></property><property><name>fs.AbstractFileSystem.oss.impl</name><value>com.aliyun.jindodata.oss.JindoOSS</value></property><property><name>fs.oss.impl</name><value>com.aliyun.jindodata.oss.JindoOssFileSystem</value></property>

配置可参考非EMR集群接入OSS-HDFS服务快速入门

配置Flume

此部分全文最关键, 请仔细看

  1. 基础配置部分, Flume配置
a1.sources = source1
a1.sinks = k1
a1.channels = c1a1.sources.source1.type = org.apache.flume.source.kafka.KafkaSource
a1.sources.source1.channels = c1
a1.sources.source1.kafka.bootstrap.servers = xxx
a1.sources.source1.kafka.topics = test
a1.sources.source1.kafka.consumer.group.id = flume-sink-group # 消费者组, 云组件需要先在管理后台创建
a1.sources.source1.kafka.consumer.auto.offset.reset = earliest # 从头消费Kafka里数据a1.sinks.k1.channel = c1
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = oss://xxx/test/%Y%m%d # 自动按天分文件夹
a1.sinks.k1.hdfs.fileType=DataStreama1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 1000

可参考使用Flume同步EMR Kafka集群的数据至OSS-HDFS服务
2. 进阶配置, 根据自己情况按需配置

a1.sinks.k1.hdfs.rollInterval = 600 # 5分钟切换一个新文件
a1.sinks.k1.hdfs.rollSize = 134217728 # 或者文件大小达到128M则切换新文件
a1.sinks.k1.hdfs.rollCount = 0 # 写入多少条数据切换新文件, 0为不限制

我这里是为了防止sink的文件过于零碎, 但因为使用的memory channel, 缓存时间过长容易丢数据
3. Flume JVM参数
默认启动时-Xmx20m, 过于小了, 加大堆内存可以直接放开flume-env.shJAVA_OPTS的注释

export JAVA_OPTS="-Xms100m -Xmx2000m -Dcom.sun.management.jmxremote"
  1. Channel问题
    如果对数据一致性要求较高, 可以把memory channel改用file channel, 请自行研究

XX启动!

几条测试命令

bin/zookeeper-server-start.sh config/zookeeper.properties # 启动zookeeper
bin/kafka-server-start.sh config/server.properties # 启动kafak服务bin/flume-ng agent --conf conf --conf-file conf/flume-conf.properties --name a1 # 启动flumebin/kafka-console-producer.sh --topic flume-test --bootstrap-server localhost:9092 # 启动一个生产者写测试数据

相关文章:

Flume采集Kafka并把数据sink到OSS

安装环境 Java环境, 略 (Flume依赖Java)Flume下载, 略Scala环境, 略 (Kafka依赖Scala)Kafak下载, 略Hadoop下载, 略 (不需要启动, 写OSS依赖) 配置Hadoop 下载JindoSDK(连接OSS依赖), 下载地址Github 解压后配置环境变量 export JINDOSDK_HOME/usr/lib/jindosdk-x.x.x expo…...

flutter,uni-app开发调试ios

一、申请ios开发者账号 二、ios开发者配置 ios 开发者需要配置的地方 https://developer.apple.com/account/resources/certificates/list Certificates&#xff08;证书&#xff09;: 作用&#xff1a; 证书用于对应用程序和开发者进行身份验证&#xff0c;确保安全性和可…...

MybatisBatchUtils功能介绍

MybatisBatchUtils 是一个 MyBatis 框架的工具类&#xff0c;主要用于简化 MyBatis 中批量操作的代码编写。该工具类封装了 MyBatis 中的批量操作方法&#xff0c;可以方便地进行批量插入、更新和删除等操作。 一般来说&#xff0c;使用 MyBatis 进行批量操作需要先设置 JDBC 驱…...

Flutter使用flutter_gen管理资源文件

pub地址&#xff1a; https://pub.dev/packages/flutter_gen 1.添加依赖 在你的pubspec.yaml文件中添加flutter_gen作为开发依赖 dependencies:build_runner:flutter_gen_runner: 2.配置pubspec.yaml 在pubspec.yaml文件中&#xff0c;配置flutter_gen的参数。指定输出路…...

vue3 setup语法糖,常用的几个:defineProps、defineEmits、defineExpose、

vue3和vue2组件之间传参的不同 <script setup> 是在单文件组件 (SFC) 中使用组合式 API 的编译时语法糖。 <script setup> 中的代码会在每次组件实例被创建的时候执行。 任何在 <script setup> 声明的顶层的绑定 (包括变量&#xff0c;函数声明&#xff0…...

JC/T 2087-2011建筑装饰用仿自然面艺术石检测

建筑装饰用仿自然面艺术石是指以硅酸盐水泥、轻质骨料为主要原料经浇筑成型的饰面装饰材料。 JC/T 2087-2011建筑装饰用仿自然面艺术石测试&#xff1a; 测试项目 测试方法 外观质量 GB/T 18601 尺寸偏差 GB/T 18601 体积密度 GB/T 9966.3 吸水率 GB/T 9966.3 压缩强…...

C语言——写一个简单函数,找两个数中最大者

#include <stdio.h>int max( int a, int b ) { return a>b ? a:b; }int main() { int a, b;printf("输入两个数:\n");scanf("%d %d", &a, &b);printf("max %d\n", max(a, b));return 0; }输出结果&#xff1a;...

机器学习中的混淆矩阵

混淆矩阵是用于评估分类模型性能的表格&#xff0c;它展示了模型在不同类别上的预测情况。对于二分类问题&#xff0c;混淆矩阵的构成如下&#xff1a; 假设有两个类别&#xff1a;正例&#xff08;Positive&#xff09;和负例&#xff08;Negative&#xff09;。 真正例&…...

QT基础实践之简易计算器

文章目录 简易计算器源码分享演示图第一步 界面设计第二步 设置槽第三步 计算功能实现 简易计算器 源码分享 链接&#xff1a;https://pan.baidu.com/s/1Jn5fJLYOZUq77eNJ916Kig 提取码&#xff1a;qwer 演示图 第一步 界面设计 这里直接用了ui界面&#xff0c;如果想要自己…...

南大通用 GBase 8s数据库级别权限

对于所有有权使用指定数据库的用户都必须赋予其数据库级别的用户权限。在GBase 8s 中&#xff0c;数据库级别的用户权限有三种&#xff0c;按权限从低到高排列依次为&#xff1a;CONNECT、RESOURCE、DBA。 1. CONNECT 这是级别最低的一种数据库级别用户权限。拥有该权限的用户…...

对话式数据需求激增,景联文科技提供高质量多轮对话数据定制采集标注服务

大模型的快速发展使得数据服务需求激增&#xff0c;产品整体处于供不应求状态。对话式数据集成为当下需求热点&#xff0c;人们对于更复杂、更真实的多轮对话数据需求不断增加&#xff0c;定制化服务占据市场需求主流。 通过对多轮对话数据的训练&#xff0c;模型可以更好地理解…...

python第1天之常识及环境安装

前言&#xff1a; 当谈到编程语言的流行度时&#xff0c;Python绝对是其中之一。Python是一种高级编程语言&#xff0c;其语法简单易懂&#xff0c;适用于各种不同的应用领域&#xff0c;包括Web开发、数据分析、人工智能等。在本文中&#xff0c;我们将探讨一些关于Pyth…...

中国高纯石英砂行业市场研究与投资前景报告(2024版)

内容简介&#xff1a; 高纯石英砂纯度高、品质好&#xff0c;生产的石英制品具有耐高温、耐腐蚀、低热膨胀性、高度绝缘性和透光性等优异的物理化学属性&#xff0c;被广泛用于光伏、电子、高端电光源、薄膜材料、国防科技等领域&#xff0c;是高端制造行业不可替代的原辅材料…...

遭到美国做空机构“灰熊”做空后,人工智能公司商汤科技股价暴跌

来源&#xff1a;猛兽财经 作者&#xff1a;猛兽财经 猛兽财经获悉&#xff0c;在遭到美国做空机构Grizzly Research&#xff08;灰熊&#xff09;指控夸大收入后&#xff0c;商汤科技的股价在周二一度下跌了9.7%。 Grizzly Research在周二发布的一份报告中称&#xff0c;商汤…...

异常数据检测 | Python实现孤立森林(IsolationForest)异常检测

孤立森林(IsolationForest)异常检测 IsolationForest[6]算法它是一种集成算法(类似于随机森林)主要用于挖掘异常(Anomaly)数据,或者说离群点挖掘,总之是在一大堆数据中,找出与其它数据的规律不太符合的数据。该算法不采样任何基于聚类或距离的方法,因此他和那些基于距离的的…...

营销互动类小游戏策划与开发

制定并开发一款营销互动小游戏需要经过一系列策划和实施步骤。以下是一个基本的流程&#xff0c;你可以根据自己的具体情况进行调整&#xff1a; 明确目标&#xff1a;确定小游戏的目标&#xff0c;是提高品牌知名度、增加销售、促进用户互动还是其他目标。 了解目标受众&…...

主机的容器化技术介绍

☞ ░ 前往老猿Python博客 ░ https://blog.csdn.net/LaoYuanPython 一、什么是容器 容器是一个标准化的单元&#xff0c;是一种轻量级、可移植的软件打包技术&#xff0c;容器将软件代码及其相关依赖打包&#xff0c;使应用程序可以在任何计算介质运行。例如开发人员在自己的…...

网络基础『发展 ‖ 协议 ‖ 传输 ‖ 地址』

&#x1f52d;个人主页&#xff1a; 北 海 &#x1f6dc;所属专栏&#xff1a; 神奇的网络世界 &#x1f4bb;操作环境&#xff1a; CentOS 7.6 阿里云远程服务器 文章目录 &#x1f324;️前言&#x1f326;️正文1.网络发展1.1.背景1.2.类型 2.网络协议2.1.什么是协议2.2.协议…...

Aapche Dubbo 不安全的 Java 反序列化 (CVE-2019-17564)

漏洞描述 Apache Dubbo 是一个高性能的、基于 Java 的开源 RPC 框架。 Apache Dubbo 支持不同的协议&#xff0c;它的 HTTP 协议处理程序是 Spring Framework 的 .org.springframework.remoting.httpinvoker.HttpInvokerServiceExporter Spring Framework 的安全警告显示&am…...

B/S软件开发架构

1.简述 1. B/S框架&#xff0c;意思是前端(Browser 浏览器&#xff0c; 小程序&#xff0c;APP&#xff0c;自己写)和服务器端(Server)组成的系统的框架结构。 2. B/S架构也可理解为web架构&#xff0c;包含前端、后端、数据库三大组成部分。 2.前端 前端开发技…...

[2025CVPR]DeepVideo-R1:基于难度感知回归GRPO的视频强化微调框架详解

突破视频大语言模型推理瓶颈,在多个视频基准上实现SOTA性能 一、核心问题与创新亮点 1.1 GRPO在视频任务中的两大挑战 ​安全措施依赖问题​ GRPO使用min和clip函数限制策略更新幅度,导致: 梯度抑制:当新旧策略差异过大时梯度消失收敛困难:策略无法充分优化# 传统GRPO的梯…...

逻辑回归:给不确定性划界的分类大师

想象你是一名医生。面对患者的检查报告&#xff08;肿瘤大小、血液指标&#xff09;&#xff0c;你需要做出一个**决定性判断**&#xff1a;恶性还是良性&#xff1f;这种“非黑即白”的抉择&#xff0c;正是**逻辑回归&#xff08;Logistic Regression&#xff09;** 的战场&a…...

在鸿蒙HarmonyOS 5中实现抖音风格的点赞功能

下面我将详细介绍如何使用HarmonyOS SDK在HarmonyOS 5中实现类似抖音的点赞功能&#xff0c;包括动画效果、数据同步和交互优化。 1. 基础点赞功能实现 1.1 创建数据模型 // VideoModel.ets export class VideoModel {id: string "";title: string ""…...

智能在线客服平台:数字化时代企业连接用户的 AI 中枢

随着互联网技术的飞速发展&#xff0c;消费者期望能够随时随地与企业进行交流。在线客服平台作为连接企业与客户的重要桥梁&#xff0c;不仅优化了客户体验&#xff0c;还提升了企业的服务效率和市场竞争力。本文将探讨在线客服平台的重要性、技术进展、实际应用&#xff0c;并…...

镜像里切换为普通用户

如果你登录远程虚拟机默认就是 root 用户&#xff0c;但你不希望用 root 权限运行 ns-3&#xff08;这是对的&#xff0c;ns3 工具会拒绝 root&#xff09;&#xff0c;你可以按以下方法创建一个 非 root 用户账号 并切换到它运行 ns-3。 一次性解决方案&#xff1a;创建非 roo…...

uniapp微信小程序视频实时流+pc端预览方案

方案类型技术实现是否免费优点缺点适用场景延迟范围开发复杂度​WebSocket图片帧​定时拍照Base64传输✅ 完全免费无需服务器 纯前端实现高延迟高流量 帧率极低个人demo测试 超低频监控500ms-2s⭐⭐​RTMP推流​TRTC/即构SDK推流❌ 付费方案 &#xff08;部分有免费额度&#x…...

解决本地部署 SmolVLM2 大语言模型运行 flash-attn 报错

出现的问题 安装 flash-attn 会一直卡在 build 那一步或者运行报错 解决办法 是因为你安装的 flash-attn 版本没有对应上&#xff0c;所以报错&#xff0c;到 https://github.com/Dao-AILab/flash-attention/releases 下载对应版本&#xff0c;cu、torch、cp 的版本一定要对…...

网站指纹识别

网站指纹识别 网站的最基本组成&#xff1a;服务器&#xff08;操作系统&#xff09;、中间件&#xff08;web容器&#xff09;、脚本语言、数据厍 为什么要了解这些&#xff1f;举个例子&#xff1a;发现了一个文件读取漏洞&#xff0c;我们需要读/etc/passwd&#xff0c;如…...

LangChain知识库管理后端接口:数据库操作详解—— 构建本地知识库系统的基础《二》

这段 Python 代码是一个完整的 知识库数据库操作模块&#xff0c;用于对本地知识库系统中的知识库进行增删改查&#xff08;CRUD&#xff09;操作。它基于 SQLAlchemy ORM 框架 和一个自定义的装饰器 with_session 实现数据库会话管理。 &#x1f4d8; 一、整体功能概述 该模块…...

在Mathematica中实现Newton-Raphson迭代的收敛时间算法(一般三次多项式)

考察一般的三次多项式&#xff0c;以r为参数&#xff1a; p[z_, r_] : z^3 (r - 1) z - r; roots[r_] : z /. Solve[p[z, r] 0, z]&#xff1b; 此多项式的根为&#xff1a; 尽管看起来这个多项式是特殊的&#xff0c;其实一般的三次多项式都是可以通过线性变换化为这个形式…...