Spark中的那些坑

• Spark中的那些坑
  • 前言
  • 讀取組態檔
  • 時區陷阱
  • 怪異的DayOfWeek
  • substring陷阱
  • IP地址解析
  • 列舉的數值
  • posexplode函數
  • 為什麼我的程式執行那麼慢？慎用Count()和Show()
  • 為什麼我的程式執行那麼慢？（2）優化、優化
  • 其它
  • 題外話

前言

在微軟內部，除了使用Cosmos+Scope處理巨量資料外，最近Hadoop+Spark的運用也逐漸多了起來。

一方面Cosmos畢竟是微軟自家的東西，很多新招來的員工以前都沒有接觸過，要熟練運用有一個熟悉的過程；而Hadoop+Spark就不一樣了，這可是巨量資料分析的標配，國外很多大學生畢業的時候已經運用相當熟練了，這樣在技術上能很快上手，減少了培訓成本；另一方面Hadoop+Spark等是開源產品，程式碼由社群維護，使用起來成本減少了不少，按照微軟一貫的做法，開源軟體外邊套個殼就可以直接上線跑了。

我們使用的Spark平臺稱之為MT，我看了目前有17個叢集，提供約104+萬個token（1個token約等於1.3個CPU核心）；我們把Spark指令碼寫好，通過上傳或者使用Python指令碼提交到伺服器上，就可以執行了。Spark對任務的設定相當地寬鬆和靈活，可以分別指定Driver和Excuter的核心數、記憶體、重試次數等，也可以使用一系列的工具對任務進行調優。目前在MT上的Haddop版本和Spark版本都是2.x，據說3.x有非常的大的改動，對效能有進一步的提升，希望微軟內部相關人員在測試後能夠儘可能早日上線，使大家早日受益。

關於Spark的運用場景我就不多說了，園內有很多大神都寫了相關介紹文章進行介紹。總的來說有兩方面，一方面是Spark結合大規模的Kafka叢集，對大量資料進行近乎實時的資料分析（QPS數千到上萬不等），稱之為DLIS。在這種場景下，一些實時資料（比如瀏覽資料、或者感測器資料）放到Kafka佇列中，由訂閱相關topic的Spark指令碼進行消費，部分資料處理後就直接送到DL（深度學習）/ML（機器學習）的模型中去了，近乎實時地、不斷地對模型進行訓練（adhoc inferencing），微軟在這方面也有其它類似的產品（比如Xap）。另一種場景就是DLOP（Deep Learning Offline Processing），可以寫一些指令碼，對每日生成的資料進行離線分析，同樣送到DL/ML的模型中去，使用另一個開源產品Airflow進行定時觸發，這種場景很類似Cosmos運用的場景，所以存在平臺互換的可能性。

正因為兩者的功能可以互換，所以越來越多的Cosmos使用者開始使用Spark。但是，進行語言切換是有使用成本的，.net平臺下的語言，換成java/Spark/scala執行的結果會有很大的不同，再此我把使用過程中碰到的Spark語言的一些坑記錄下來，以供大家參考，避免在今後的工作中踩坑。

讀取組態檔

這是一份普通的組態檔 Sample.config

input {
  path = "/aa/bb/cc";
  format = "parquet";
  saveMode = "overwrite";
}

當Spark讀取這份設定後，不會丟擲任何錯誤，但是程式中始終無法讀取該input檔案，提示路徑不對。

解決方法：

java或者Spark的組態檔在每行末尾不需要新增任何結束符號，包括「;」、「,」等，像這個例子中，實際讀取到的path是"/aa/bb/cc;",自然是不存在的路徑。
正確的組態檔格式

input {
  path = "/aa/bb/cc"
  format = "parquet"
  saveMode = "overwrite"
}

時區陷阱

我們在國際化的程式中，需要對儲存的時間進行處理，儲存為統一的UTC時間戳。當Spark讀取到UTC時間，使用hour函數讀取該日期的小時值，在不同時區的電腦上獲得的值不同。
範例：

Dataset<Row> data = data.withColumn("Hour", hour(col("SomeTime")));

我們假設某一行SomeTime的值為 2022-10-23 5:17:14 (UTC時間)，那麼在美國的群集上執行，hour函數取得的值為22（太平洋夏令時時間），在香港的伺服器上執行，那麼獲取到的hour值為13（UTC+8）。

解決方法

在對UTC時間進行計算前先根據本地時區轉換為UTC標準時間，例：

在美國的伺服器上執行：

Dataset<Row> data = data.withColumn("Hour", hour(to_utc_timestamp(col("SomeTime"), "America/Los_Angeles")));
或
Dataset<Row> data = data.withColumn("Hour", hour(to_utc_timestamp(col("SomeTime"), "GMT-7")));

在香港的伺服器上執行：

Dataset<Row> data = data.withColumn("Hour", hour(to_utc_timestamp(col("SomeTime"), "China/Beijing")));
或
Dataset<Row> data = data.withColumn("Hour", hour(to_utc_timestamp(col("SomeTime"), "GMT+8")));

*使用.net處理時區沒有問題，不管電腦在那裡執行，不會把UTC-0的時間轉換為本地時間再進行運算。

怪異的DayOfWeek

一個不難理解的日期函數，在不同的語言上，竟得出不一樣的結果。

C#

    var date = DateTime.Parse("2022-11-11");
    Console.WriteLine(date.DayOfWeek);

結果是：Friday

java:

import java.time.DayOfWeek;
import java.time.LocalDate;
...
    LocalDate date = LocalDate.of(2022, 11, 11);
    DayOfWeek week = DayOfWeek.from(date);
    System.out.println(week); 

結果是：FRIDAY

Spark：

Dataset<Row> data = data.withColumn("Hour", dayofweek(col("SomeTime")));

結果是：6

問題原因

Spark用了個很怪的序列儲存Dayofweek的值，它的序列從Sunday開始，到Saturday結束，返回的值分別是1,2,3,4,5,6,7，Saturday的值是7，我也是無語了。

解決方法

如果要與java或者.net返回的dayofweek值保持一致，寫一個UDF（User Define Function）來進行轉換，比如：

    /**
     * Get the C# DayOfWeek Name by Spark Value
     */
    UserDefinedFunction udfGetDayOfWeekNameBySparkValue = udf((Integer dayOfWeek) -> {
        if(dayOfWeek > 7 || dayOfWeek < 1 ){
            return "0";
        }
        String[] DayOfWeeks = (String[]) Arrays.asList("0", "Sunday", "Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday").toArray();
        return DayOfWeeks[dayOfWeek];
    }, StringType);

substring陷阱

在Spark中使用substring函數獲取字串中的一部分，比如說有一個字串的日期格式是20221204（YYYYMMDD），那麼我需要分別獲取他的年、月、日，使用如下程式碼

Dataset<Row> data = data.withColumn("YEAR", substring(col("SomeTime"), 0, 4));
    .withColumn("MONTH", substring(col("SomeTime"), 4, 2));
    .withColumn("DAY", substring(col("SomeTime"), 6, 2));

結果是：

YEAR --> 2022 --> 正確！
MONTH --> 21 --> 錯誤！
DAY --> 20 --> 錯誤！

問題原因

在Spark，substring的下標從1開始，這點和c#或者java有很大的不同。

解決方法

把初始下標換成1，那麼就能取得正確的值，如下：

Dataset<Row> data = data.withColumn("YEAR", substring(col("SomeTime"), 1, 4));
    .withColumn("MONTH", substring(col("SomeTime"), 5, 2));
    .withColumn("DAY", substring(col("SomeTime"), 7, 2));

這點在Spark的官方檔案 functions (Spark 3.3.1 JavaDoc) 中有一行小字說明 。

但是，如果是從第一個字元開始取值，那麼使用0或者1會取得相同的結果，也就是

Dataset<Row> data = data.withColumn("YEAR", substring(col("SomeTime"), 0, 4));
和
Dataset<Row> data = data.withColumn("YEAR", substring(col("SomeTime"), 1, 4));

取到的值居然是相同的！

但，如果在Spark中使用陣列，那麼下標仍然是從0開始！暈~

IP地址解析

在工作中，我們經常要對IP地址進行解析，使用C#程式碼可以方便地對IPV4地址進行校驗和劃分網段：

    string IpAddress = "202.96.205.133";
    string[] octets = IpAddress.Split(".");
    Console.WriteLine(octets[0]);

輸出：202

我們把它轉換成java程式碼：

    String IpAddress = "202.96.205.133";
    String[] octets = IpAddress.split(".");
    System.out.println(octets[0]);

執行，它居然報錯了：java.lang.ArrayIndexOutOfBoundsException: Index 0 out of bounds for length 0

問題原因

java的split函數接收的引數其實是正規表示式，一些符號需要做跳脫，比如 * ^ : | . \ 等。

解決方法

在「.」前加上雙斜槓 \\ 即可。

    String IpAddress = "202.96.205.133";
    String[] octets = IpAddress.split("\\.");
    System.out.println(octets[0]);

這樣就得到和C#一樣的結果。

列舉的數值

有這麼一個列舉

C# & java

    public enum Color {
        Red,
        Green,
        Blue,
        Gray
        ........
    }

有時候我們需要獲取列舉的數值，比如Red所代表的數值，使用C#可以方便地取得列舉相應的值：

C#

    Color.Red.ToString("D")
    或者
    (int)Color.Red

輸出： 0

但是，如果你用java，卻無論如何獲取不到列舉所代表的數值。

解決方法

在java中，要獲取列舉所代表的數值，需要在列舉定義中新增一些額外的程式碼。

    public enum Color {
        Red(0),
        Green(1),
        Blue(2),
        Gray(3)
        ........
    }

    private final int value;

    private LocationLevelType(int value){
        this.value = value;
    }

    public int getValue() {
        return value;
    }

這樣使用如下程式碼就可以獲取每個列舉所代表的數值了：

    Color.Red.value
    或者
    Color.valueOf("Red").value

輸出：0

posexplode函數

有時候我們需要對一個陣列型別的列進行拆分，獲得每一項的值，單獨生成行值和索引，這時候可以使用posexplode函數。在Spark的官方檔案 functions (Spark 3.3.1 JavaDoc) 中，明確說明該函數從2.1.0版本就得到支援；但是，我們在Spark2.12環境中，該函數仍然不見蹤影。

解決方法

使用selectExpr的方法進行替代，如下所示：

Dataset<Row> data = data.selectExpr("*", "posexplode(some_array) as (index, content)")

為什麼我的程式執行那麼慢？慎用Count()和Show()

「我的程式執行了好長時間，還停留在初始步驟，我已經分配了足夠的資源，輸入的資料量也不是特別大啊？」

問題原因

檢查一下你的程式碼，是不是加了很多埋點，使用Count()和Show()來計算Dataset的行數？
在Spark中，Count()和Show()是個非常消耗資源的操作，他會把該中間過程的所有資料都計算出來，然後給出一個統計值，或者若干行的輸出。在不快取資料的情況下，加一個Count()或者Show()，都會使執行時間成倍增長。

解決方法

去掉那些埋點，原先幾天都跑不完的程式碼幾小時就跑完了。
在生產環境中執行的程式碼萬萬不建議使用Count()和Show()函數。實在不得已要用，也要適當增加計算資源，做好job失敗和成倍的執行時間的準備。

為什麼我的程式執行那麼慢？（2）優化、優化

「什麼，這麼簡單job用了500個token，跑了一天還跑不完？那麼多節點重試出錯？趕緊去優化！」

問題原因

Spark是個非常靈活的資料處理平臺，可以執行各種符合規範的java程式碼，限制明顯比Scope、Sql少很多。它的機制是把任務分解到各個容器中去執行，直到有輸出的時候再計算，並沒有統一快取的中間過程，所以執行效率取決於你寫程式的好壞，它能做的優化有限。
比方說，原始表有25TB資料，255列，你在程式碼裡Count()了一下、選擇某些列進行和其他表join、再從原始表中根據某些條件篩選一些資料，和join的資料union。那麼，一共需要掃描三遍原始資料表，讀取75TB的資料量，執行時間驚人。

解決方法

選擇只需要用到的列（一般不會超過50列），快取，進行Count()（小心這一步往往會失敗），再按照正常步驟join和union，這樣讀取一遍，快取5TB資料；如果不需要Count，還可以事先快取和過濾。

其它

使用java程式碼解析windows下生成的文字檔案要注意換行符是\r\n，不是\n，否則從第二行開始，每一行的開頭都會多一個空格。在對物件進行BASE64編碼的時候也要選擇合適的類，否則和C#編譯出來的會不同。

題外話

微軟的Cosmos是內部使用的巨量資料平臺，沒有對外開放（參見我的聊聊我在微軟外服巨量資料分析部門的工作經歷及一些個人見解），.net缺少類似Spark這樣的巨量資料平臺和生態，目前微軟在這方面的解決方案是Azure上的Datalake（資料湖）+ Synapase SQl；希望.net今後能夠向開源巨量資料發展方向發力，從而打造起自己的生態，和java一爭高下！

Intel最近釋出了針對資料中心、AI學習的旗艦級志強晶片Intel Xeon Max，它具有56個核心、64GB HBM2e記憶體，在一些橫向測試中比競品和上一代產品功耗降低、效能提升2~4倍不等。我不知道根據美國晶片法禁令，這樣的晶片能不能進口，如果不能進口的話，我國在巨量資料分析、AI領域方面的差距將進一步與國際拉大。希望我們國家能夠自立自強，早日生產出能夠運用於商業化的超算晶片，使得華為這樣的公司不再處處受排擠。

微軟外服工作札記系列
①聊聊我在微軟外服巨量資料分析部門的工作經歷及一些個人見解
②聊聊微軟的知識管理服務平臺和一些程式設計風格
③視窗函數的介紹
④Spark中的那些坑
⑤微軟內部的知識圖譜Satori介紹
⑥聊聊我認識的那些印度人