2. 程式人生 > >Cris 玩轉大資料系列之訊息佇列神器 Kafka

Cris 玩轉大資料系列之訊息佇列神器 Kafka

阿新 • • 發佈：2018-11-30

Cris 玩轉大資料系列之訊息佇列神器 Kafka

Author：Cris

文章目錄

1. Kafka 概述

1.1 訊息佇列

先來看看訊息佇列的實現原理圖（以 Kafka 為例）

① 點對點模式

一對一，消費者主動拉取資料，訊息收到後訊息清除

點對點模型通常是一個基於拉取或者輪詢的訊息傳送模型，這種模型從佇列中請求資訊，而不是將訊息推送到客戶端。這個模型的特點是傳送到佇列的訊息被一個且只有一個接收者接收處理，即使有多個訊息監聽者也是如此

② 釋出/訂閱模式

一對多，資料生產後，推送給所有訂閱者

釋出訂閱模型則是一個基於推送的訊息傳送模型。釋出訂閱模型可以有多種不同的訂閱者，臨時訂閱者只在主動監聽主題時才接收訊息，而持久訂閱者則監聽主題的所有訊息，即使當前訂閱者不可用，處於離線狀態

1.2 為什麼需要訊息佇列？

  1. 解耦

    允許你獨立的擴充套件或修改兩邊的處理過程，只要確保它們遵守同樣的介面約束

  2. 冗餘

    訊息佇列把資料進行持久化直到它們已經被完全處理，通過這一方式規避了資料丟失風險。許多訊息佇列所採用的"插入-獲取-刪除"正規化中，在把一個訊息從佇列中刪除之前，需要你的處理系統明確的指出該訊息已經被處理完畢，從而確保你的資料被安全的儲存直到你使用完畢

  3. 擴充套件性

    因為訊息佇列解耦了你的處理過程，所以增大訊息入隊和處理的頻率是很容易的，只要另外增加處理過程即可

  4. 靈活性 & 峰值處理能力

    在訪問量劇增的情況下，應用仍然需要繼續發揮作用，但是這樣的突發流量並不常見。如果為以能處理這類峰值訪問為標準來投入資源隨時待命無疑是巨大的浪費。使用訊息佇列能夠使關鍵元件頂住突發的訪問壓力，而不會因為突發的超負荷的請求而完全崩潰

  5. 可恢復性

    系統的一部分元件失效時，不會影響到整個系統。訊息佇列降低了程序間的耦合度，所以即使一個處理訊息的程序掛掉，加入佇列中的訊息仍然可以在系統恢復後被處理

  6. 順序保證

    在大多使用場景下，資料處理的順序都很重要。大部分訊息佇列本來就是排序的，並且能保證資料會按照特定的順序來處理。（Kafka保證一個Partition內的訊息的有序性）

  7. 緩衝

    有助於控制和優化資料流經過系統的速度，解決生產訊息和消費訊息的處理速度不一致的情況

  8. 非同步通訊

    很多時候，使用者不想也不需要立即處理訊息。訊息佇列提供了非同步處理機制，允許使用者把一個訊息放入佇列，但並不立即處理它。想向佇列中放入多少訊息就放多少，然後在需要的時候再去處理它們

1.3 什麼是 Kafka？

在流式計算中，Kafka一般用來快取資料，Storm通過消費Kafka的資料進行計算

1）Apache Kafka是一個開源訊息系統，由Scala寫成。是由Apache軟體基金會開發的一個開源訊息系統專案

2）Kafka最初是由LinkedIn公司開發，並於2011年初開源。2012年10月從Apache Incubator畢業。該專案的目標是為處理實時資料提供一個統一、高通量、低等待的平臺

3）**Kafka是一個分散式訊息佇列。**Kafka對訊息儲存時根據Topic進行歸類，傳送訊息者稱為Producer，訊息接受者稱為Consumer，此外kafka叢集有多個kafka例項組成，每個例項(server)稱為broker

4）無論是kafka叢集，還是consumer都依賴於zookeeper叢集儲存一些meta資訊，來保證系統可用性

整體架構圖

詳細架構圖

1）Producer ：訊息生產者，就是向kafka broker發訊息的客戶端；

2）Consumer ：訊息消費者，向kafka broker取訊息的客戶端；

3）Topic ：可以簡單理解為一個佇列；

4） Consumer Group （CG）：這是kafka用來實現一個topic訊息的廣播（發給所有的consumer）和單播（發給任意一個consumer）的手段。一個topic可以有多個CG。topic的訊息會複製（不是真的複製，是概念上的）到所有的CG，但每個partion只會把訊息發給該CG中的一個consumer。如果需要實現廣播，只要每個consumer有一個獨立的CG就可以了。要實現單播只要所有的consumer在同一個CG。用CG還可以將consumer進行自由的分組而不需要多次傳送訊息到不同的topic；

5）Broker ：一臺kafka伺服器就是一個broker。一個叢集由多個broker組成。一個broker可以容納多個topic；

6）Partition：為了實現擴充套件性，一個非常大的topic可以分佈到多個broker（即伺服器）上，一個topic可以分為多個partition，每個partition是一個有序的佇列。partition中的每條訊息都會被分配一個有序的id（offset）。kafka只保證按一個partition中的順序將訊息發給consumer，不保證一個topic的整體（多個partition間）的順序；

7）Offset：kafka的儲存檔案都是按照offset.log來命名，用offset做名字的好處是方便查詢。例如你想找位於2049的位置，只要找到2048.log的檔案即可。當然the first offset就是00000000000.log

2. Kafka 叢集部署

瞭解完訊息佇列和 Kafka 的基礎知識，接下來就來搭建一個微型 Kafka 叢集，通過對這個微型 Kafka 叢集的操作來快速上手 Kafka

2.1 Kafka 叢集規劃

hadoop101 hadoop102 hadoop103
zk zk zk
Kafka Kafka Kafka

2.2 jar 包下載

http://kafka.apache.org/downloads.html

2.3 Kafka叢集部署

1）解壓安裝包

[[email protected] software]$ tar -zxvf kafka_2.11-0.11.0.0.tgz -C /opt/module/

2）修改解壓後的檔名稱

[[email protected] module]$ mv kafka_2.11-0.11.0.0/ kafka

3）在/opt/module/kafka目錄下建立logs資料夾

[[email protected] kafka]$ mkdir logs

4）修改配置檔案

[[email protected] kafka]$ cd config/
[[email protected] config]$ vim server.properties

對應下面內容進行修改：

#broker的全域性唯一編號，不能重複
broker.id=1

#刪除topic功能使能
delete.topic.enable=true

#處理網路請求的執行緒數量
num.network.threads=3

#用來處理磁碟IO的現成數量
num.io.threads=8

#傳送套接字的緩衝區大小
socket.send.buffer.bytes=102400
#接收套接字的緩衝區大小
socket.receive.buffer.bytes=102400
#請求套接字的緩衝區大小
socket.request.max.bytes=104857600

#kafka執行日誌存放的路徑	
log.dirs=/opt/module/kafka/logs
#topic在當前broker上的分割槽個數
num.partitions=1
#用來恢復和清理data下資料的執行緒數量
num.recovery.threads.per.data.dir=1
#segment檔案保留的最長時間，超時將被刪除
log.retention.hours=168
#配置連線Zookeeper叢集地址
zookeeper.connect=hadoop101:2181,hadoop102:2181,hadoop103:2181

5）配置環境變數

vim /etc/profile

6）分發安裝包

xsync kafka/

注意配置其他機子的環境變數

7）分別在hadoop102和hadoop103上修改配置檔案/opt/module/kafka/config/server.properties中的broker.id=2、broker.id=3

​ 注：broker.id不得重複

8）啟動叢集

這裡 Cris 寫了一個群起指令碼

寫指令碼之前先複習一下 Shell 的基礎知識

1、由單引號括起來的字元都作為普通字元出現。特殊字元用單引號括起來以後，也會失去原有意義，而只作為普通字元解釋

2、由雙引號括起來的字元，除$（美元符號）、\（反斜槓）、’（單引號）、和”（雙引號）這幾個字元仍是特殊字元並保留其特殊功能外，其餘字元仍作為普通字元對待

具體的參考文章

然後指令碼具體內容如下：

cat /home/cris/bin/clusters | while read line
do
{
	echo "開始啟動-->"$line
	ssh $line "source /etc/profile;nohup ${KAFKA_HOME}/bin/kafka-server-start.sh ${KAFKA_HOME}/config/server.properties >/dev/null 2>&1 &" 
}&
wait
done
echo '***Kafka叢集啟動完成***'

其中 cluster 的內容如下：

hadoop101
hadoop102
hadoop103

改指令碼具體的解釋這裡不過多解釋，但是需要注意的是：執行該指令碼需要 hadoop101，hadoop102，hadoop103 實現免密登入，具體實現流程請參考 Cris 之前的 Hadoop 筆記

關於 & 符號：後臺執行 shell 命令

關於 /dev/null ：輸出重定向常用的黑洞

關於 wait：等待 shell 命令執行

關於 nohup：終端關閉不會影響已經開啟的程序

具體 Linux 命令參考文章：

nohup 參考

wait 參考

/dev/null 參考

Linux 特殊符號命令集合

其他參考文章一

群起指令碼參考一

群起指令碼參考二

然後是群停指令碼

cat /home/cris/bin/clusters | while read line
do
{
	echo "開始停止-->"$line
	ssh $line "source /etc/profile;nohup ${KAFKA_HOME}/bin/kafka-server-stop.sh >/dev/null 2>&1 &" 
}&
wait
done
echo '***Kafka叢集正在從 Zookeeper 上登出停止，請等待一段時間再使用 jpsall 命令***'

其中 jpsall 指令用來檢視所用機子上的 Java 程序

效果如下：

注意：啟動 Kafka 叢集之前請先啟動 Zookeeper 叢集

2.4 Kafka 命令列操作

1）檢視當前伺服器中的所有topic

[[email protected] bin]$ kafka-topics.sh --zookeeper hadoop101:2181 --list

2）建立topic

[[email protected] kafka]$ kafka-topics.sh --zookeeper hadoop101:2181 
--create --topic first --partitions 1 --replication-factor 3

選項說明：

–topic 定義topic名

–replication-factor 定義副本數

–partitions 定義分割槽數

3）刪除topic

kafka-topics.sh --zookeeper hadoop101:2181 
--delete --topic first

需要server.properties中設定delete.topic.enable=true否則只是標記刪除或者直接重啟

演示圖如下：

4）傳送訊息

[[email protected] bin]$  kafka-console-producer.sh --broker-list hadoop101:9092 --topic first

5）消費訊息

[[email protected] ~]$ kafka-console-consumer.sh --zookeeper hadoop102:2181 --from-beginning --topic first

–from-beginning：會把first主題中以往所有的資料都讀取出來。根據業務場景選擇是否增加該配置

6）檢視某個Topic的詳情

[[email protected] ~]$ kafka-topics.sh --zookeeper hadoop101:2181 --describe --topic first

3. Kafka 工作流程分析

3.1 寫入方式

producer採用推（push）模式將訊息釋出到broker，每條訊息都被追加（append）到分割槽（patition）中，屬於順序寫磁碟（順序寫磁碟效率比隨機寫記憶體要高，保障kafka吞吐率）

3.2 分割槽（Partition）

訊息傳送時都被髮送到一個topic，其本質就是一個目錄，而topic是由一些Partition Logs(分割槽日誌)組成，其組織結構如下圖所示：

我們可以看到，每個Partition中的訊息都是有序的，生產的訊息被不斷追加到Partition log上，其中的每一個訊息都被賦予了一個唯一的offset值

1）分割槽的原因

（1）方便在叢集中擴充套件，每個Partition可以通過調整以適應它所在的機器，而一個topic又可以有多個Partition組成，因此整個叢集就可以適應任意大小的資料了；

（2）可以提高併發，因為可以以Partition為單位讀寫了。

2）分割槽的原則

（1）指定了patition，則直接使用；

（2）未指定patition但指定key，通過對key的value進行hash出一個patition；

（3）patition和key都未指定，使用輪詢選出一個patition。

Kafka 原始碼定義的預設分割槽類參考如下：

DefaultPartitioner類
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        if (keyBytes == null) {
            int nextValue = nextValue(topic);
            List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
            if (availablePartitions.size() > 0) {
                int part = Utils.toPositive(nextValue) % availablePartitions.size();
                return availablePartitions.get(part).partition();
            } else {
                // no partitions are available, give a non-available partition
                return Utils.toPositive(nextValue) % numPartitions;
            }
        } else {
            // hash the keyBytes to choose a partition
            return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
        }
    }

3.3 副本（Replication）

同一個partition可能會有多個replication（對應 server.properties 配置中的 default.replication.factor=N）。沒有replication的情況下，一旦broker 宕機，其上所有 patition 的資料都不可被消費，同時producer也不能再將資料存於其上的patition。引入replication之後，同一個partition可能會有多個replication，而這時需要在這些replication之間選出一個leader，producer和consumer只與這個leader互動，其它replication作為follower從leader 中複製資料

3.4 寫入流程（待修改）

1）producer先從zookeeper的 "/brokers/…/state"節點找到該partition的leader

2）producer將訊息傳送給該leader

3）leader將訊息寫入本地log

4）followers從leader pull訊息，寫入本地log後向leader傳送ACK

5）leader收到所有ISR中的replication的ACK後，增加HW（high watermark，最後commit 的offset）並向producer傳送ACK

3.5 Broker 儲存訊息

① 儲存方式

物理上把topic分成一個或多個patition（對應 server.properties 中的num.partitions=3配置），每個patition物理上對應一個資料夾（該資料夾儲存該patition的所有訊息和索引檔案），如下：

② 儲存策略

無論訊息是否被消費，kafka都會保留所有訊息。有兩種策略可以刪除舊資料：

1）基於時間：log.retention.hours=168

2）基於大小：log.retention.bytes=1073741824

需要注意的是，因為Kafka讀取特定訊息的時間複雜度為O(1)，即與檔案大小無關，所以這裡刪除過期檔案與提高 Kafka 效能無關

3.6 Zookeeper儲存結構

注意：producer不在zk中註冊，消費者在zk中註冊。

3.7 Kafka消費過程分析

kafka提供了兩套consumer API：高階Consumer API和低階Consumer API

① 高階 API

1）高階API優點

高階API 寫起來簡單

不需要自行去管理offset，系統通過zookeeper自行管理。

不需要管理分割槽，副本等情況，.系統自動管理。

消費者斷線會自動根據上一次記錄在zookeeper中的offset去接著獲取資料（預設設定1分鐘更新一下zookeeper中存的offset）

可以使用group來區分對同一個topic 的不同程式訪問分離開來（不同的group記錄不同的offset，這樣不同程式讀取同一個topic才不會因為offset互相影響）

2）高階API缺點

不能自行控制offset（對於某些特殊需求來說）

不能細化控制如分割槽、副本、zk等

② 低階 API

1）低階 API 優點

能夠讓開發者自己控制offset，想從哪裡讀取就從哪裡讀取。

自行控制連線分割槽，對分割槽自定義進行負載均衡

對zookeeper的依賴性降低（如：offset不一定非要靠zk儲存，自行儲存offset即可，比如存在檔案或者記憶體中）

2）低階API缺點

太過複雜，需要自行控制offset，連線哪個分割槽，找到分割槽leader 等。

③ 消費者組

消費者是以consumer group消費者組的方式工作，由一個或者多個消費者組成一個組，共同消費一個topic。每個分割槽在同一時間只能由group中的一個消費者讀取，但是多個group可以同時消費這個partition。在圖中，有一個由三個消費者組成的group，有一個消費者讀取主題中的兩個分割槽，另外兩個分別讀取一個分割槽。某個消費者讀取某個分割槽，也可以叫做某個消費者是某個分割槽的擁有者。

在這種情況下，消費者可以通過水平擴充套件的方式同時讀取大量的訊息。另外，如果一個消費者失敗了，那麼其他的group成員會自動負載均衡讀取之前失敗的消費者讀取的分割槽

一句話：分組中的消費者可以讀取 topic 中的多個分割槽，但是 topic 中的一個分割槽不能被多個同組的消費者同時讀取

④ 消費方式

consumer採用pull（拉）模式從broker中讀取資料。

push（推）模式很難適應消費速率不同的消費者，因為訊息傳送速率是由broker決定的。它的目標是儘可能以最快速度傳遞訊息，但是這樣很容易造成consumer來不及處理訊息，典型的表現就是拒絕服務以及網路擁塞。而pull模式則可以根據consumer的消費能力以適當的速率消費訊息

對於Kafka而言，pull模式更合適，它可簡化broker的設計，consumer可自主控制消費訊息的速率，同時consumer可以自己控制消費方式——即可批量消費也可逐條消費，同時還能選擇不同的提交方式從而實現不同的傳輸語義。

pull模式不足之處是，如果kafka沒有資料，消費者可能會陷入迴圈中，一直等待資料到達。為了避免這種情況，我們在我們的拉請求中有引數，允許消費者請求在等待資料到達的“長輪詢”中進行阻塞（並且可選地等待到給定的位元組數，以確保大的傳輸大小）

⑤ 消費者組案例

1）需求：測試同一個消費者組中的消費者，同一時刻只能有一個消費者消費一個 topic 的分割槽資訊

2）案例實操

  1. 在hadoop102、hadoop103上修改/opt/module/kafka/config/consumer.properties配置檔案中的group.id屬性為任意組名

    [[email protected] ~]$ vim /opt/module/kafka/config/consumer.properties

    hadoop103 同理

  2. 先啟動 Kafka 叢集，然後分別啟動 hadoop102 和 hadoop103 的消費者客戶端

    [[email protected] kafka]$ kafka-console-consumer.sh --zookeeper hadoop102:2181 --topic first --consumer.config config/consumer.properties

[[email protected] kafka]$ kafka-console-consumer.sh --zookeeper hadoop102:2181 --topic first --consumer.config config/consumer.properties

[[email protected] bin]$ kafka-console-producer.sh --broker-list hadoop102:9092 --topic first

    可以發現，hadoop101 生產者傳送的資料只有同一組的 hadoop102 取到了，而沒有被 hadoop103 重複獲取

4. Kafka API實戰

4.1 環境準備

1）啟動zk和kafka叢集，在kafka叢集中開啟一個消費者

這裡 Cris 建立兩個別名，來方便的從 Terminal 開啟 Kafka 消費者和生產者

vim ~/.bashrc

alias kafka_consumer_topic="kafka-console-consumer.sh --zookeeper hadoop101:2181 --topic $1"

alias kafka_producer_topic="kafka-console-producer.sh --broker-list hadoop101:9092 --topic $1"

然後執行 source ~/.bashrc 即可

下次想要啟動消費者或者生產者輸入類似下面命令即可：

kafka_consumer_topic first
kafka_producer_topic first

然後啟動 Zookeeper 叢集

接著啟動 Kafka 叢集

接著啟動一個消費者客戶端

2）匯入 pom 依賴

<dependencies>
    <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>0.11.0.0</version>
    </dependency>
    <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka_2.12</artifactId>
        <version>0.11.0.0</version>
    </dependency>
</dependencies>

4.2 Kafka 生產者 API

① 過時的建立生產者 API（瞭解）

public class OldProducer {

	@SuppressWarnings("deprecation")
	public static void main(String[] args) {
		
		Properties properties = new Properties();
		properties.put("metadata.broker.list", "hadoop102:9092");
		properties.put("request.required.acks", "1");
		properties.put("serializer.class", "kafka.serializer.StringEncoder");
		
		Producer<Integer, String> producer = new Producer<Integer,String>(new ProducerConfig(properties));
		
		KeyedMessage<Integer, String> message = new KeyedMessage<Integer, String>("first", "hello world");
		producer.send(message );
	}
}

② 新建立生產者 API

/**
 * 新建立 Producer 的 API，儘量使用 Kafka 官方配置類的常量
 *
 * @author cris
 * @version 1.0
 **/
public class MyProducer {
    public static void main(String[] args) {
        Properties prop = new Properties();

        // Kafka服務端的主機名和埠號
        prop.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop101:9092");
        // 等待所有副本節點的應答(最嚴格的資料儲存方式，效率也最低，還可以取值 0 或者 1)
        prop.setProperty(ProducerConfig.ACKS_CONFIG, "all");
        // 訊息傳送失敗最大嘗試次數
        prop.put(ProducerConfig.RETRIES_CONFIG, 0);
        // 一批訊息處理大小
        prop.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        // 請求延時
        prop.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        // 傳送快取區記憶體大小
        prop.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        // key序列化
        prop.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // value序列化
        prop.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(prop);
        producer.send(new ProducerRecord<>("first", "simida"));

        producer.close();
    }
}

執行效果如下

③ 建立生產者帶回調函式（新API）

其實就是在生產者傳送訊息後得到 broker 確定收到訊息的呼叫的函式，注意的是，不帶回調函式和帶回調函式的 send 方法都是非同步方法，只有呼叫 send().get() 才是同步方法，瞭解即可

/**
 * 新建立 Producer 的 API，帶有回撥函式
 *
 * @author cris
 * @version 1.0
 **/
@SuppressWarnings("Duplicates")
public class MyCallbackProducer {

    private static final Logger LOGGER = LoggerFactory.getLogger(String.valueOf(MyCallbackProducer.class));

    public static void main(String[] args) {
        Properties prop = new Properties();

        // Kafka服務端的主機名和埠號
        prop.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop101:9092");
        // 等待所有副本節點的應答(最嚴格的資料儲存方式，效率也最低，還可以取值 0 或者 1)
        prop.setProperty(ProducerConfig.ACKS_CONFIG, "all");
        // 訊息傳送失敗最大嘗試次數
        prop.put(ProducerConfig.RETRIES_CONFIG, 0);
        // 一批訊息處理大小
        prop.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        // 請求延時
        prop.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        // 傳送快取區記憶體大小
        prop.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        // key序列化
        prop.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // value序列化
        prop.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(prop);

        producer.send(new ProducerRecord<>("first", "curry"),((recordMetadata, e) -> {
            if (recordMetadata != null) {
                System.out.println("recordMetadata.topic() = " + recordMetadata.topic());
                System.out.println("recordMetadata.offset() = " + recordMetadata.offset());
                System.out.println("recordMetadata.partition() = " + recordMetadata.partition());
            }else {
                LOGGER.info("metadata is null！");
            }
        }));

        producer.close();
    }
}

④ 自定義分割槽生產者

0）需求：將所有資料儲存到topic的第0號分割槽上

1）定義一個類實現Partitione

相關推薦

Cris 玩轉大資料系列之訊息佇列神器 Kafka

Cris 玩轉大資料系列之訊息佇列神器 Kafka Author：Cris 文章目錄 Cris 玩轉大資料系列之訊息佇列神器 Kafka Author：Cris 1. Kafka 概述

玩轉大資料系列之二：資料分析與處理

經過了資料採集和同步之後，就可以在阿里雲上進行資料分析和處理，來玩轉您的資料了。本文向您介紹在阿里雲大資料各產品中，以及各產品之間怎樣來完成您的資料處理和資料分析。 MaxCompute 基於MaxCompute的大資料計算（MaxCompute + RDS） 使用MaxCompute分析IP

玩轉大資料系列之三：資料報表與展示

經過了資料採集與資料同步、資料分析和處理，我們應該考慮將處理好的資料做成報表或者大屏展示給老闆們看，以便老闆們可以更加精準地做出戰略決策，為業務的發展指明方向。 提到資料報表，不得不說說Quick BI。Quick BI提供海量資料實時線上分析服務，支援拖拽式操作、提供了豐富的視覺化效果，可以幫助您輕鬆自如

玩轉大資料系列之四：搜尋服務

搜尋服務是阿里雲產品非常重要的組成部分，也承載了阿里巴巴集團的全部主要的搜尋任務。這裡的搜尋服務主要包含兩個產品：阿里雲Elasticsearch和開放搜尋OpenSearch。 本文將介紹如何玩轉阿里雲搜尋的大資料服務。 阿里雲Elasticsearch 阿里雲Elasticsearch，提供基於開源

玩轉大資料系列之一：資料採集與同步

資料的採集和同步，是先將資料從裝置、或者本地資料來源採集、同步到阿里雲上，然後在阿里雲上對資料進行分析和處理，最終完成您的業務要求。本文向您介紹阿里雲各產品的資料採集和同步的操作實戰文章，您可以根據您使用阿里雲產品，檢視相應的文件教程。 關於資料採集，DataWorks專門有一個模組叫做資料整合，是阿里巴巴

大資料系列之Flume(1.6)+kafka 整合

相關文章：關於Flume 的 一些核心概念：元件名稱    功能介紹Agent代理使用JVM 執行Flume。每臺機器執行一個agent，但是可以在一個agent中包含多個sources和sinks。Client客戶端生產資料，執行在一個獨立的執行緒。Source源從Client收集資料，傳遞給Channel

玩轉大資料！Hadoop在大資料的地位如此之高？

馬雲曾說過：“未來最大的資源就是資料，不參與大資料十年後一定會後悔！” 馬雲 Hadoop 最早誕生於 2006 年，並在 2008 年成為 Apache 頂級專案，起初，Hadoop在中國基本沒有人注意到它，直到2011年開始，中國進入大資料風起雲湧的時代，因

玩轉大資料之——通過java操作office/wps的 EXCEL

操作Office文件，要用到apache公司的一個工具包: poi-3.7-20101029.jar poi外掛的使用方式類似dom技術 不同版本： 接下來進行演示功能: 1.用純Java技術實現建立一個名為a.xls的檔案(工作薄)， 在其中新

你愛的小米是怎樣玩轉大資料的？大咖揭祕小米大資料整合架構演化之路

小米有眾多的智慧終端和裝置，資料規模非常大，對於資料採集和大資料整合提出了非常高的要求。此次演講主要介紹小米大資料整合解決方案，主要包括小米資料流平臺的架構演化，整個鏈路的資料質量監控，資料流生態的構建思路，最後會介紹典型的應用場景、未來的規劃和思考。 分享大綱： 1、問題與挑戰 2、資

大資料系列之分散式釋出訂閱訊息系統Kafka（一）Kafka簡介，組成，叢集安裝

1.Kafka簡介   Kafka如同JMS(Java Message Service)一樣，是一箇中間件，在異構系統間通訊，為不同的系統之間提供服務。我們知道JMS通過佇列（一對一）與主題（一對多）兩種形式提供服務，而Kafka則通過主題（topic），來給一組消費者提供

大資料系列之分散式釋出訂閱訊息系統Kafka（四）Kafka與Flume的3種整合

前面我們已經介紹了Flume，現在我們將Kafka與Flume整合 先看一下Flume的結構組成：            我們可以發現，將Flume與Kafka進行整合無非3種情況，Flume作為生產者——Sink輸出到Kafka，Flume作為消費者——Source接

【大資料新手上路】“零基礎”系列課程--如何通過大資料開發套件Data IDE玩轉大資料

本教程是一個大資料開發套件Data IDE零基礎教程，通過Data IDE將多源異構的資料集匯入雲端MaxCompute，進行計算、分析；然後通過Data IDE系統排程定時處理，最終，通過Quick BI實現企業T-1天的業務報表及分析；

Flume+Kafka雙劍合璧玩轉大資料平臺日誌採集

概述 大資料平臺每天會產生大量的日誌，處理這些日誌需要特定的日誌系統。 一般而言，這些系統需要具有以下特徵： 構建應用系統和分析系統的橋樑，並將它們之間的關聯解耦； 支援近實時的線上分析系統和類似於Hadoop之類的離線分析系統； 具有高可擴充套件性。即：當資料量增加時，可以通過增加節點

大資料系列之——hive（七、hive詳解及應用）

目錄 2.HQL 一、HIVE概述 1.Hadoop分散式計算遇到的問題 MapReduce只能用java開發(也支援其他語言，但是不是主流)需要對Hadoop的底層原理 api比較瞭解才能順暢的開發出分散式的處

大資料系列之hive（八、hive內建函式全解）

1.內建運算子1.1關係運算符 運算子 型別 說明 A = B 所有原始型別 如果A

Python 玩轉大資料 Mapreduce開發 wordcount

一 介紹 MapReduce 是一種分散式程式設計模型，用於處理大規模的資料。使用者主要通過指定一個 map 函式和一個 reduce 函式來處理一個基於key/value pair的資料集合，輸出中間的基於key/value pair的資料集合；然後 再建立一

玩轉大資料視覺化的幾個必會工具，你認識幾個？

俗話說的好，逆水行舟，不進則退，在快速發展的今天，緊跟時代的步伐，抓住時代的脈搏，才能助力企業激流勇進，搶佔先機。作為助力企業經營決策的大資料視覺化應用，對於的企業發展起著至關重要的作用，但是如何利用大資料視覺化，如何做好大資料視覺化，今天小編就給大家介紹幾款實用的工具，幫助大家在大資料的海洋裡開發

大資料系列之實時計算Spark（十三）機器學習

1.機器學習簡介 機器學習可能是當下最火的話題了。之前我們所做的一些工作，比如說java開發，安卓等等，其實無非就是在來回的寫方法，呼叫方法，而機器學習說的通俗一點可能就是找函式。要知道，我們現在面對的是巨大的資料量，對於這麼多的資料量，我們不太可能找到一個描述資料的方法或

大資料系列之Hadoop知識整理（七）MapReduce的核心之Shuffle詳解

1.MapReduce的核心之shuffle詳解 上一篇中我們介紹了MapReduce是什麼，以及MapReduce的執行過程，其中在執行過程中主要分為Map端與Reducer端，MapReduce計算模型主要完成了對映與化簡，在這其中，有一個最重要的過程那就是其核心——s

玩轉大數據系列之三：數據報表與展示

效果 工程 過濾 業務 數據同步 應用 不同 ref 運營 經過了數據采集與數據同步、數據分析和處理，我們應該考慮將處理好的數據做成報表或者大屏展示給老板們看，以便老板們可以更加精準地做出戰略決策，為業務的發展指明方向。 提到數據報表，不得不說說Quick BI。Quick