消息隊(duì)列后端流處理【免費(fèi)下載鏈接】pulsarApache Pulsar - distributed pub-sub messaging system項(xiàng)目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar點(diǎn)擊查看免費(fèi)下載HDFS2 Sink 是 Apache Pulsar 內(nèi)置的 IO 連接器之一負(fù)責(zé)從 Pulsar Topic 拉取消息并將消息以文件形式持久化到 Hadoop HDFSHadoop 2.x 生態(tài)中是Pulsar 消息 → 數(shù)據(jù)湖/離線分析存儲(chǔ)鏈路上最常見的落庫(kù)手段之一。本文以 site2/website-next/docs/io-hdfs2-sink.md 為骨架結(jié)合 pulsar-io/hdfs2 模塊的源碼與測(cè)試完整講解其全部配置屬性、校驗(yàn)規(guī)則、JSON/YAML 配置模板以及底層寫入與確認(rèn)ack機(jī)制幫助你在實(shí)際集群中正確部署并調(diào)優(yōu)該連接器。HDFS2 Sink 連接器是什么HDFS2 Sink 是一個(gè) Pulsar IO Sink 組件核心職責(zé)一句話概括從 Pulsar Topic 拉取消息并將其持久化到 HDFS 文件中。它適合如下場(chǎng)景將實(shí)時(shí)消息流持續(xù)歸檔到 HDFS供 Hive、Spark、Presto 等離線/批量計(jì)算引擎消費(fèi)將 Pulsar 作為統(tǒng)一事件總線Sink 作為數(shù)據(jù)出口之一把數(shù)據(jù)搬運(yùn)到 HDFS需要與 Hadoop 2.x 體系含 Kerberos 認(rèn)證打通的存算分離架構(gòu)。從倉(cāng)庫(kù)結(jié)構(gòu)看該連接器位于 pulsar-io/hdfs2其模塊名為pulsar-io-hdfs2構(gòu)建時(shí)依賴hadoop-client2.8.5見 pulsar-io/hdfs2/pom.xml因此它面向 Hadoop 2.x 生態(tài)同倉(cāng)庫(kù)中的 pulsar-io/hdfs3 則面向 Hadoop 3.x。數(shù)據(jù)寫入的核心機(jī)制源碼級(jí)解讀HDFS2 Sink 不是簡(jiǎn)單地把每條消息直接寫盤而是通過緩沖隊(duì)列 后臺(tái)同步線程實(shí)現(xiàn)批量落盤與確認(rèn)。整條鏈路由以下類協(xié)作完成HdfsSinkConfig.java負(fù)責(zé)加載并校驗(yàn)配置HdfsAbstractSink.javaSink 主體負(fù)責(zé)連接 HDFS、創(chuàng)建文件、啟動(dòng)同步線程HdfsSyncThread.java后臺(tái)線程周期調(diào)用hsync()刷盤并 ack 已落盤記錄HdfsAbstractTextFileSink.java及seq包下的序列文件實(shí)現(xiàn)負(fù)責(zé)實(shí)際寫入格式AbstractHdfsConnector.javaHDFS 連接與 Kerberos 認(rèn)證的底層封裝。啟動(dòng)流程open()在 HdfsAbstractSink.java#L57-L69 中open()依次執(zhí)行通過HdfsSinkConfig.load(config)解析配置支持 Map 與 YAML 文件兩種入口調(diào)用hdfsSinkConfig.validate()做參數(shù)合法性校驗(yàn)用maxPendingRecords作為容量創(chuàng)建LinkedBlockingQueueRecordV即未確認(rèn)記錄隊(duì)列若配置了subdirectoryPattern則編譯對(duì)應(yīng)的DateTimeFormatter連接 HDFSconnectToHdfs()、創(chuàng)建文件寫入器createWriter()、啟動(dòng)同步線程launchSyncThread()。文件路徑與文件命名目標(biāo)文件路徑在 HdfsAbstractSink.java#L100-L118 的getPath()中生成規(guī)則為{directory}/{subdirectoryPattern 格式化的當(dāng)前時(shí)間}/ {filenamePrefix}-{System.currentTimeMillis()}{擴(kuò)展名}其中擴(kuò)展名的優(yōu)先級(jí)是fileExtension配置優(yōu)先若未配置fileExtension但設(shè)置了compression則使用壓縮編解碼器的默認(rèn)擴(kuò)展名如 GZIP 的.gz。這也解釋了文檔中filenamePrefix 使topicA產(chǎn)生名為topicA-...的文件的描述文件名由filenamePrefix - 時(shí)間戳拼接而成。寫入、刷盤與確認(rèn)寫入以文本格式為例HdfsAbstractTextFileSink.java#L56-L69 將每條記錄record.getValue().toString()寫入OutputStreamWriter若配置了separator非默認(rèn)\u0000空字符則在記錄后追加分隔符寫入成功后記錄放入未確認(rèn)隊(duì)列寫入失敗則調(diào)用record.fail()。同步與確認(rèn)HdfsSyncThread.java#L46-L78 的后臺(tái)線程每隔syncInterval毫秒執(zhí)行一次先對(duì) HDFS 輸出流調(diào)用hsync()強(qiáng)制刷盤再把隊(duì)列中的記錄逐一ack()。當(dāng)syncInterval為 0 時(shí)線程會(huì)盡快循環(huán)刷盤close()時(shí)調(diào)用halt()做最后一次刷盤并確認(rèn)全部待確認(rèn)記錄。背壓語義未確認(rèn)隊(duì)列容量即maxPendingRecords。當(dāng)隊(duì)列滿時(shí)put()會(huì)阻塞從而對(duì)上游形成天然背壓保證內(nèi)存中的未確認(rèn)記錄數(shù)不超過閾值。這正是文檔中設(shè)置為 1 時(shí)每條記錄先落盤再 ack最多一次內(nèi)存緩沖、設(shè)置為較大值時(shí)允許批量緩沖后統(tǒng)一刷盤的底層原因整體遵循at-least-once至少一次投遞語義。連接與安全認(rèn)證AbstractHdfsConnector.java#L64-L97 的resetHDFSResources()負(fù)責(zé)初始化 HadoopConfiguration與FileSystem按逗號(hào)切分hdfsConfigResources逐個(gè)addResource加載配置資源需能被類路徑找到否則拋出IOException關(guān)閉FileSystem緩存fs.scheme.impl.disable.cachetrue避免重配置后無法生效若集群?jiǎn)⒂昧?KerberosSecurityUtil.isSecurityEnabled則以kerberosUserPrincipal keytab登錄并執(zhí)行doAs獲取 FileSystem否則走simple認(rèn)證。配置屬性全解析HDFS2 Sink 的配置由 HdfsSinkConfig.javaSink 特有屬性與 AbstractHdfsConfig.javaHDFS 通用屬性共同承載全部屬性如下表名稱類型必填默認(rèn)值說明hdfsConfigResourcesString是None包含 Hadoop 文件系統(tǒng)配置的一個(gè)文件或逗號(hào)分隔的文件列表。示例core-site.xmlhdfs-site.xmldirectoryString是NoneHDFS 中讀取或?qū)懭胛募哪夸?。encodingString否None文件的字符編碼。示例UTF-8ASCIIcompressionCompression否NoneHDFS 上壓縮/解壓文件所使用的壓縮編解碼器可選值如下BZIP2DEFLATEGZIPLZ4SNAPPYkerberosUserPrincipalString否None用于認(rèn)證的 Kerberos 用戶主體principal。keytabString否None用于認(rèn)證的 Kerberos keytab 文件的完整路徑。filenamePrefixString是compression為None時(shí)NoneHDFS 目錄內(nèi)創(chuàng)建文件的前綴。示例取值為topicA時(shí)將生成名為topicA-...的文件。fileExtensionString是None寫入 HDFS 文件的擴(kuò)展名。示例.txt.seqseparatorchar否None文本文件中用于分隔記錄的字符。若未設(shè)置所有記錄的內(nèi)容將首尾相連拼接為一個(gè)連續(xù)字節(jié)數(shù)組。syncIntervallong否0調(diào)用 flush 將數(shù)據(jù)寫入 HDFS 磁盤的間隔毫秒。maxPendingRecordsint否Integer.MAX_VALUEack 之前允許在內(nèi)存中持有的最大記錄數(shù)。設(shè)為 1 時(shí)每條記錄在 ack 前都會(huì)先落盤設(shè)為大值時(shí)允許先緩沖多條記錄再統(tǒng)一刷盤。subdirectoryPatternString否None與 Sink 創(chuàng)建時(shí)間關(guān)聯(lián)的子目錄。該模式是directory子目錄的格式化模式。模式語法見 DateTimeFormatter該鏈接為 Oracle 官方 JDK 文檔供查閱模式語法。關(guān)鍵參數(shù)深度說明hdfsConfigResources與directory二者是 AbstractHdfsConfig.java#L66-L75 中強(qiáng)校驗(yàn)的必填項(xiàng)任一缺失都會(huì)拋出Required property not set.。hdfsConfigResources指向的文件會(huì)被config.addResource()加載到 HadoopConfiguration中必須位于連接器的類路徑上。compression與fileExtension的聯(lián)動(dòng)HdfsSinkConfig.java#L94-L99 的校驗(yàn)邏輯是當(dāng)fileExtension為空且compression也為空時(shí)才報(bào)錯(cuò)即只要設(shè)置了壓縮fileExtension可省略將由CompressionCodec.getDefaultExtension()補(bǔ)全如 GZIP →.gz。同時(shí) AbstractHdfsConnector.java#L182-L191 通過CompressionCodecFactory.getCodecByName()按名稱解析編解碼器枚舉定義見 Compression.javaBZIP2, DEFLATE, GZIP, LZ4, SNAPPY。kerberosUserPrincipal與keytab二者必須成對(duì)出現(xiàn)——只配置其一都會(huì)拋出Values for both kerberosUserPrincipal keytab are required.見 AbstractHdfsConfig.java#L71-L74。syncInterval與maxPendingRecordssyncInterval不能為負(fù)數(shù)maxPendingRecords必須為正整數(shù)。它們共同決定吞吐優(yōu)先還是延遲/可靠性優(yōu)先追求低延遲逐條落盤可設(shè)maxPendingRecords1追求高吞吐可調(diào)大該值并配合合理的syncInterval。subdirectoryPattern模式會(huì)先經(jīng) HdfsSinkConfig.java#L109-L115 用固定時(shí)間LocalDateTime.of(2020, 1, 1, 12, 0)做格式合法性校驗(yàn)非法模式會(huì)在啟動(dòng)階段直接報(bào)錯(cuò)運(yùn)行時(shí)則由 HdfsAbstractSink.java#L110-L112 用LocalDateTime.now()生成實(shí)際子目錄。常見的按天歸檔寫法為yyyy-MM-dd。encoding若未配置則回退到 JVM 默認(rèn)字符集見 AbstractHdfsConnector.java#L177-L180?？缂翰渴饡r(shí)建議顯式指定如UTF-8避免因運(yùn)行環(huán)境不同導(dǎo)致亂碼。separator未配置時(shí)默認(rèn)值為 Java 空字符\u0000寫入時(shí)不做分隔見 HdfsAbstractTextFileSink.java#L61-L63配置后每條記錄后追加該字符方便下游按分隔符切分。配置示例使用 HDFS2 Sink 連接器前需先通過以下任一方式創(chuàng)建配置文件以下兩例均完整復(fù)現(xiàn)自官方文檔可直接套用。JSON 示例{ configs: { hdfsConfigResources: core-site.xml, directory: /foo/bar, filenamePrefix: prefix, fileExtension: .log, compression: SNAPPY, subdirectoryPattern: yyyy-MM-dd } }YAML 示例configs: hdfsConfigResources: core-site.xml directory: /foo/bar filenamePrefix: prefix fileExtension: .log compression: SNAPPY subdirectoryPattern: yyyy-MM-dd以上示例對(duì)應(yīng)的解析結(jié)果hdfsConfigResourcescore-site.xml、directory/foo/bar、filenamePrefixprefix、compressionSNAPPY、subdirectoryPatternyyyy-MM-dd在測(cè)試類 HdfsSinkConfigTests.java#L39-L48 與 #L51-L66 中被逐一斷言驗(yàn)證可直接作為配置正確性的參照?？焖偕鲜植渴?HDFS2 Sink前置條件可訪問的 Hadoop 2.x 集群本模塊依賴hadoop-client2.8.5將core-site.xml、hdfs-site.xml等 Hadoop 配置資源放入連接器可訪問的類路徑可放置于 NAR 包內(nèi)或通過extraDependenciesDir掛載因?yàn)榈讓油ㄟ^config.addResource()按資源名加載若 HDFS 開啟 Kerberos需準(zhǔn)備 principal 與 keytab 文件路徑目標(biāo) HDFS 目錄已存在或運(yùn)行賬戶有創(chuàng)建權(quán)限文件不存在時(shí)連接器會(huì)調(diào)用fs.create(path)存在時(shí)追加寫入見 HdfsAbstractSink.java#L95。創(chuàng)建 Sink使用 Pulsar Admin CLI 的 sinks 命令即可創(chuàng)建archive 指向構(gòu)建出的 HDFS2 Sink NAR 包可用mvn package從 pulsar-io/hdfs2 構(gòu)建bin/pulsar-admin sinks create \ --tenant public \ --namespace default \ --name hdfs2-sink \ --sink-type hdfs2 \ --archive /path/to/pulsar-io-hdfs2.nar \ --inputs input-topic \ --sink-config-file /path/to/sink-config.yaml創(chuàng)建后Sink 實(shí)例會(huì)以 Pulsar Function 運(yùn)行時(shí)的方式運(yùn)行消費(fèi)input-topic上的消息并按前述流程寫入 HDFS。暫?；騽h除可分別使用bin/pulsar-admin sinks pause與bin/pulsar-admin sinks delete。具體參數(shù)說明可查閱倉(cāng)庫(kù) pulsar-client-tools 模塊中 sinks 相關(guān)命令實(shí)現(xiàn)。常見配置組合與調(diào)優(yōu)建議歸檔場(chǎng)景追求吞吐maxPendingRecords調(diào)大如 10000syncInterval設(shè)為幾百到幾千毫秒批量刷盤換取更高吞吐文件擴(kuò)展名建議配合compression如 SNAPPY壓縮落盤節(jié)省 HDFS 空間。嚴(yán)格可靠性場(chǎng)景maxPendingRecords1每條消息先落盤再 acksyncInterval保持較小值。代價(jià)是吞吐顯著下降僅建議在低流量關(guān)鍵鏈路使用。按天分目錄歸檔設(shè)置subdirectoryPattern: yyyy-MM-ddHDFS 上會(huì)生成/foo/bar/2026-09-23/prefix-timestamp.log這樣的層級(jí)結(jié)構(gòu)配合directory統(tǒng)一管理生命周期。多條消息混寫不設(shè)置separator時(shí)記錄首尾相接下游解析需按固定長(zhǎng)度設(shè)置separator如\n后可按行切分更利于文本類下游工具直接處理。測(cè)試與驗(yàn)證倉(cāng)庫(kù)在 pulsar-io/hdfs2/src/test 下提供了完整的配置與端到端測(cè)試可用于驗(yàn)證你的配置理解HdfsSinkConfigTests.java覆蓋 YAML/Map 加載、必填項(xiàng)缺失、非法壓縮編解碼器、負(fù) syncInterval、非正 maxPendingRecords、Kerberos 參數(shù)不成對(duì)等全部校驗(yàn)分支AbstractHdfsSinkTest.java 及text/seq包下的 HdfsStringSinkTests.java、HdfsTextSinkTests.java、HdfsSequentialSinkTests.java 則驗(yàn)證了實(shí)際寫入行為?？偨Y(jié)HDFS2 Sink 是 Apache Pulsar 連接 HDFS 2.x 生態(tài)的標(biāo)準(zhǔn)出口它以緩沖隊(duì)列 后臺(tái)hsync刷盤 批量 ack的機(jī)制在吞吐與可靠性之間提供了maxPendingRecords、syncInterval兩個(gè)核心旋鈕通過compression、fileExtension、subdirectoryPattern等參數(shù)可靈活控制落盤文件的格式、壓縮與目錄布局同時(shí)借助 Hadoop 原生配置資源與 Kerberos 支持無縫融入企業(yè)級(jí)安全集群。理解 HdfsSinkConfig.java 的校驗(yàn)邏輯與 HdfsAbstractSink.java 的寫入流程是正確配置與排障的關(guān)鍵起點(diǎn)。贊分享消息隊(duì)列后端流處理【免費(fèi)下載鏈接】pulsarApache Pulsar - distributed pub-sub messaging system項(xiàng)目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar點(diǎn)擊查看免費(fèi)下載相關(guān)推薦Apache Pulsar Kafka Sink Connector 實(shí)戰(zhàn)指南將 Pulsar Topic 消息橋接到 KafkaApache Pulsar Kafka Sink Connector 實(shí)戰(zhàn)指南將 Pulsar Topic 消息橋接到 Kafka Kafka Sink Co消息隊(duì)列后端流處理Apache Pulsar Solr Sink Connector 配置與源碼剖析將 Topic 消息持久化到 Solr CollectionApache Pulsar Solr Sink Connector 配置與源碼剖析將 Topic 消息持久化到 Solr Collection Solr si消息隊(duì)列后端流處理Apache Pulsar Redis Sink Connector 完全指南將 Topic 消息實(shí)時(shí)寫入 RedisApache Pulsar Redis Sink Connector 完全指南將 Topic 消息實(shí)時(shí)寫入 Redis 本篇技術(shù)指南以 Apache Puls消息隊(duì)列后端流處理創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考