chinesefreesexvideos高潮,欧美极品少妇性运交,久久久国产一区二区三区,99久久婷婷国产综合精品,成人国产一区二区三区

超長文警告！本文7000字，含架構(gòu)圖和各種解決方案的嘗試，以及詳細(xì)代碼。，最后還有電子書和各種分享ppt下載，請(qǐng)?jiān)趙ifi下觀看。土豪隨意

我是鳴宇淳，一個(gè)大數(shù)據(jù)架構(gòu)師。今天給大家分享一下我在傳統(tǒng)金融業(yè)務(wù)場(chǎng)景中，用Flink的特性，幫助業(yè)務(wù)解決實(shí)際問題的案例。

Flink 有很好的支持事件時(shí)間、有狀態(tài)的計(jì)算、可以靈活設(shè)置窗口等特點(diǎn)，受到業(yè)界的廣泛追捧。我所在的公司并不是典型的互聯(lián)網(wǎng)公司，而是偏傳統(tǒng)的金融公司，我也想探索一下怎么能利用 Flink 這個(gè)好的工具來服務(wù)公司的業(yè)務(wù)，所以對(duì) Flink 的應(yīng)用場(chǎng)景做了一番嘗試。

    下面我將探索的過程記錄下來，一是分享下自己的經(jīng)歷，二是希望跟同行交流，還望輕拍。

我公司業(yè)務(wù)場(chǎng)景跟市面上大部分 Flink 使用場(chǎng)景不太一樣的地方在于以下幾點(diǎn)：

1. 表單類數(shù)據(jù)比較多

要處理的數(shù)據(jù)里，日志類型的數(shù)據(jù)很少，基本上都是表單類的業(yè)務(wù)數(shù)據(jù)，這類數(shù)據(jù)會(huì)經(jīng)常變化。

2. 數(shù)據(jù)更新周期長

一條業(yè)務(wù)數(shù)據(jù)可能需要幾天或者幾個(gè)月后，才有一條更新操作，所以不能更好地利用 Flink 的窗口，而 Flink 更適合于那種插入后很快就會(huì)有更新的情況。

3. 維表數(shù)據(jù)變化時(shí)，歷史統(tǒng)計(jì)結(jié)果數(shù)據(jù)也要跟著變

維表數(shù)據(jù)如果變了，統(tǒng)計(jì)結(jié)果數(shù)據(jù)都要跟著變，這也是傳統(tǒng)業(yè)務(wù)中常見要求。

4. 數(shù)據(jù)準(zhǔn)確度要求非常高

因?yàn)槭墙鹑跇I(yè)務(wù)，多數(shù)場(chǎng)景都和金錢相關(guān)，所以要求計(jì)算出的數(shù)據(jù)準(zhǔn)確。

以上幾條都是我們傳統(tǒng)金融業(yè)務(wù)中比較特別的場(chǎng)景。

另外，由于現(xiàn)有人員多數(shù)是精通 SQL 的開發(fā)人員，而精通 Java、Scala 開發(fā)的并不多，所以決定盡量不使用 Flink API 的方式開發(fā) Flink 實(shí)時(shí)計(jì)算。所以自研了一個(gè)基于 Flink 的實(shí)時(shí)計(jì)算平臺(tái)，主要功能有使用 Flink SQL 創(chuàng)建各類 Source和Sink的 Flink 表、創(chuàng)建 Flink 任務(wù)、提交 Flink 任務(wù)到 Yarn 上執(zhí)行、任務(wù)監(jiān)控等，另外也可以基于 Flink API 開發(fā)，將 jar 表上傳到實(shí)時(shí)計(jì)算平臺(tái)上執(zhí)行。

以上是大致背景。接下來是我們具體的探索和實(shí)踐過程。

我們根據(jù)業(yè)務(wù)需求，決定把Flink作為實(shí)時(shí)計(jì)算平臺(tái)。前面通過Kafka收集和傳輸數(shù)據(jù)，通過Flink計(jì)算平臺(tái)處理好之后，把數(shù)據(jù)扔到MySQL、ES、HBase等存儲(chǔ)落地，或者繼續(xù)扔到Kafka中等待下一步計(jì)算。

以此，我們就可以畫出實(shí)時(shí)數(shù)據(jù)處理的整體流程：

1. 數(shù)據(jù)來源有兩類：一類是業(yè)務(wù)數(shù)據(jù)，絕大部分來源于 MySQL，這部分?jǐn)?shù)據(jù)通過 Canal 將 Binlog 數(shù)據(jù)實(shí)時(shí)寫入 Kafka中；另外一類是少量的日志數(shù)據(jù)，這些數(shù)據(jù)通過 Flume 實(shí)時(shí)采集后，也寫入 Kafka中。

2. 在實(shí)時(shí)計(jì)算平臺(tái)上，通過 Flink SQL 定義好 source 表、sink 表，指定 Kafka 來源和要寫入的目標(biāo)庫。

3. 或者是將開發(fā)的 Flink 程序打成 jar 包后上傳到實(shí)時(shí)計(jì)算平臺(tái)上。

4. 在實(shí)時(shí)計(jì)算平臺(tái)上創(chuàng)建任務(wù)，運(yùn)行各種業(yè)務(wù)計(jì)算。

5. 如果計(jì)算結(jié)果為最終數(shù)據(jù)，就 sink 到 MySQL、ES、HBase 等存儲(chǔ)中用于展示，如果計(jì)算結(jié)果數(shù)據(jù)還需要進(jìn)行下一步計(jì)算，那么就 sink 到 Kafka 中。

0. 需求

業(yè)務(wù)背景：以金融業(yè)務(wù)中常見的還款計(jì)劃表為例。比如有一張客戶還款計(jì)劃表，里面每一條數(shù)據(jù)是一期要還款的記錄，每個(gè)單號(hào)有多行數(shù)據(jù)，主要字段有：主鍵、單號(hào)、期號(hào)、計(jì)劃還款日期、實(shí)際還款日期、應(yīng)還金額、是否逾期、還款狀態(tài)等。

具體業(yè)務(wù)需求是：通過實(shí)時(shí)計(jì)算，統(tǒng)計(jì)出每個(gè)單號(hào)的首期月供、首期還款日期、首次逾期日期、當(dāng)前期號(hào)、實(shí)際已還金額、剩余金額等統(tǒng)計(jì)數(shù)據(jù)，每個(gè)單號(hào)一行結(jié)果。

大體需求如下：

1.實(shí)現(xiàn)

這個(gè)需求使用 Flink Java API 來實(shí)現(xiàn)，程序大體流程如下：

簡單來說：

1. 一共設(shè)計(jì)了兩張 HBase 表：保存所有明細(xì)數(shù)據(jù)的 HBase 表、保存實(shí)時(shí)統(tǒng)計(jì)結(jié)果的 HBase 表。

2. 對(duì)于 Kafka 傳遞過來的每一條 binlog 日志，不管是 insert 還是 update，都在窗口內(nèi)保存到 Flink 的狀態(tài)里，并進(jìn)行合并，只保留每個(gè)單子最新一條 binlog 日志的全量字段值。

3. 當(dāng)窗口關(guān)閉的時(shí)候，從明細(xì) HBase 表中讀取相應(yīng)單號(hào)的數(shù)據(jù)，和 Flink 狀態(tài)中的最新數(shù)據(jù)進(jìn)行合并，并將合并后的最新數(shù)據(jù)更新到 HBase 明細(xì)表中，完成實(shí)時(shí)同步。

4. 同時(shí)對(duì)這個(gè)窗口 Flink 狀態(tài)中涉及到的單號(hào)，使用最新的數(shù)據(jù)進(jìn)行統(tǒng)計(jì)，然后將統(tǒng)計(jì)結(jié)果 Sink 到 HBase 結(jié)果表中。

代碼總體邏輯如下：

    public static void main(String[] args) throws Exception {

        //定義一個(gè)Flink Stream執(zhí)行環(huán)境對(duì)象

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //設(shè)置checkpoint

        env.enableCheckpointing(600 * 1000);

        //設(shè)置狀態(tài)后端

        String chkpointPath = "hdfs://ha/xxx/checkpoint";

        StateBackend backend = new RocksDBStateBackend(chkpointPath);

        env.setStateBackend(backend);

        //設(shè)置事件時(shí)間        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //Kafka的ip和要消費(fèi)的topic

        String kafkaIPs = "********";

        String topic = "plan_topic";

        //Kafka設(shè)置

        Properties props = new Properties();

        props.setProperty("bootstrap.servers", kafkaIPs);

        props.setProperty("group.id", "group.cyb.1");

        FlinkKafkaConsumer>> consumer =

                new FlinkKafkaConsumer>>(topic, new KafkaDeserializationScheme(), props);

        //設(shè)置水位線        AssignerWithPeriodicWatermarks>> watermarks =

                new PeriodicTsAndWmarks<>();

        consumer.assignTimestampsAndWatermarks(watermarks);

        consumer.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor>>(Time.seconds(5)) {

            @Override

            public long extractTimestamp(KafkaData> element) {

                return element.getTs();

            }

        });

        //傳入?yún)?shù)，指定從哪個(gè)時(shí)間戳開始消費(fèi)Kafka

        long startTs = Long.parseLong(args[0]);

        //設(shè)置從指定時(shí)間戳開始消費(fèi)

        consumer.setStartFromTimestamp(startTs);

        SingleOutputStreamOperator ds = env

                //指定Source

                .addSource(consumer)

                //篩選數(shù)據(jù)，只是保留指定表的數(shù)據(jù)

                .filter(f -> f.getTable().toLowerCase().equals("plan_cust") && !f.getIsDdl())

                //因?yàn)閗afka里的一條數(shù)據(jù)可能包括多條insert、update、delete記錄，所以binlog日志是一對(duì)多的結(jié)構(gòu)

                //這里進(jìn)行打平，一條kafka日志輸出多條記錄

                .flatMap(new DataFlatMap())

                //根據(jù)單號(hào)進(jìn)行分組

                .keyBy(RepayPlanCustOut::getApply_no)

                //定義窗口分配器，根據(jù)EventTime進(jìn)行滾動(dòng)。

                .window(TumblingEventTimeWindows.of(Time.seconds(20)))

                //增量聚合

                //.aggregate(new CustAggregateFunction())

                //延遲指定時(shí)間后，窗口才釋放

                //.allowedLateness(Time.seconds(5))

                //對(duì)每個(gè)窗口內(nèi)的所有元素執(zhí)行函數(shù)操作

                .process(new CustProcessWindowFun())

                //指定并行度

                .setParallelism(15)

                //因?yàn)槭褂昧薼amada表達(dá)式，所以需要指定返回類型

                .returns(new TypeHint() {

                });

        //輸出到自定義的Sink

        ds.writeUsingOutputFormat(new SinkOutFormat())

                //指定并行度

                .setParallelism(15);

        //提交執(zhí)行Flink任務(wù)，指定任務(wù)名稱

        env.execute("planMain-1");

    }

2. 數(shù)據(jù)初始化和數(shù)據(jù)測(cè)試

數(shù)據(jù)初始化和數(shù)據(jù)測(cè)試的流程如下：

數(shù)據(jù)初始化

不像常見 Flink 實(shí)時(shí)計(jì)算場(chǎng)景那樣，上線時(shí)不用考慮歷史數(shù)據(jù)問題，我這個(gè)業(yè)務(wù)場(chǎng)景中，在程序首次上線或者迭代上線的時(shí)候，需要將歷史數(shù)據(jù)也做初始化，初始化工作包括將歷史數(shù)據(jù)同步到 HBase 明細(xì)表中、將統(tǒng)計(jì)結(jié)果數(shù)據(jù)初始化到 HBase 統(tǒng)計(jì)結(jié)果表中。

數(shù)據(jù)初始化采取的方案是：使用 Hive 中的歷史數(shù)據(jù)同步到 HBase 表中。

一是創(chuàng)建一個(gè) HBase Hive 外部表，將明細(xì)數(shù)據(jù) Insert 到 HBase 明細(xì)表中；

二是使用 Hive 寫同樣的 統(tǒng)計(jì)腳本，將統(tǒng)計(jì)結(jié)果 Insert 到 HBase 統(tǒng)計(jì)結(jié)果表中；

數(shù)據(jù)測(cè)試

對(duì)于統(tǒng)計(jì)結(jié)果的測(cè)試，使用一個(gè) Python 腳本，分別讀取 MySQL 業(yè)務(wù)備庫中的數(shù)據(jù)、HBase 統(tǒng)計(jì)結(jié)果中的數(shù)據(jù)，然后進(jìn)行自動(dòng)對(duì)比。

以上就是對(duì)于單表實(shí)時(shí)計(jì)算的介紹，下面來介紹更為復(fù)雜的多表關(guān)聯(lián)場(chǎng)景。

0. 需求

存儲(chǔ)在 MySQL 業(yè)務(wù)庫中的好幾張業(yè)務(wù)表，包括：合同主表、**表、產(chǎn)品方案表等等，基于這些表，需要實(shí)時(shí)進(jìn)行關(guān)聯(lián)，得到的結(jié)果如下圖所示：

1. 探索：方案一

首先想到的最簡單的方式就是這樣，涉及到的每個(gè)從表都使用 Flink 做實(shí)時(shí)同步到 HBase 中，關(guān)聯(lián)鍵為 RowKey。然后 Flink 對(duì)于主表的每一條 binlog 日志進(jìn)行讀取后，關(guān)聯(lián) HBase 中的每個(gè)從表，得到的結(jié)果寫入 HBase 結(jié)果表或者傳遞給下游 Kafka。

這個(gè)方案優(yōu)點(diǎn)是邏輯簡單，缺點(diǎn)是維表數(shù)據(jù)有延遲、要?jiǎng)?chuàng)建很多 HBase表，尤其是有多種關(guān)聯(lián)方式時(shí)。

2. 探索：方案二

方案一中使用 HBase 當(dāng)維表是因?yàn)楫?dāng)時(shí)使用的 Flink 1.5 中不支持 MySQL 當(dāng)維表，后來實(shí)時(shí)計(jì)算平臺(tái)的 Flink 升級(jí)到了 1.10后，可以使用 MySQL 當(dāng)維表，拋棄了 HBase 那就簡單多了，程序處理流程變?yōu)榱松厦娴臉幼印?/span>

(1) 只有主表實(shí)時(shí)接入到Topic。

(2) Flink讀取主表Topic數(shù)據(jù)，實(shí)時(shí)關(guān)聯(lián)MySQL中維表。

(3) 將關(guān)聯(lián)結(jié)果寫入MySQL結(jié)果表中。

(4) 可以同時(shí)將關(guān)聯(lián)結(jié)果寫入Kafka，用做下一步計(jì)算。

Flink 實(shí)時(shí)計(jì)算程序直接讀取 MySQL 主表和維表進(jìn)行計(jì)算，這樣維表的更新沒有了延遲。

但是后來發(fā)現(xiàn)有個(gè)問題，就是當(dāng)維表數(shù)據(jù)更新時(shí)，不能觸發(fā)計(jì)算，不能更新結(jié)果。

3. 探索：方案三

由于方案二中有個(gè)致命問題，就是由于 Flink 是事件驅(qū)動(dòng)的，只有主表更新時(shí)才觸發(fā)計(jì)算，而維表數(shù)據(jù)變化時(shí)，也需要更新歷史老數(shù)據(jù)（這個(gè)估計(jì)是傳統(tǒng)業(yè)務(wù)場(chǎng)景中的特有的要求），所以形成了方案三，如上圖所示：

(1) 所有表都實(shí)時(shí)接入到 Topic 。

(2) 每個(gè)表的接入都要觸發(fā)計(jì)算，用結(jié)果表當(dāng)維表。

(3) 將關(guān)聯(lián)結(jié)果寫入 MySQL 結(jié)果表中。

(4) 可以同時(shí)將關(guān)聯(lián)結(jié)果寫入 Kafka ，用做下一步計(jì)算。

這樣主表、維表數(shù)據(jù)更新都可以觸發(fā)計(jì)算，但是還有個(gè)缺點(diǎn)，就是結(jié)果表每次都是對(duì)全部字段讀取后，再全部字段更新，效率低。

4. 探索：方案四

怎么解決方案三中效率低的問題呢，然后就有了方案四:

(1) 所有表都實(shí)時(shí)接入到 Topic 。

(2) 每個(gè)表的接入都要觸發(fā)計(jì)算，用結(jié)果表當(dāng)維表關(guān)聯(lián)，更新結(jié)果表。只更新各自相關(guān)字段。

(3) 將關(guān)聯(lián)結(jié)果寫入 MySQL 結(jié)果表中。

這樣主表、維表數(shù)據(jù)更新都可以觸發(fā)計(jì)算，提高了執(zhí)行效率，但是呢也是有問題的，就是步驟較多，程序復(fù)雜。

Flink總體使用體感還是很舒服的，速度很快，跟其他組件對(duì)接也很順暢，尤其是對(duì)于事件時(shí)間、有狀態(tài)的計(jì)算等特性非常好用。

但是我們的一些場(chǎng)景也不太符合Flink的特性，比如數(shù)據(jù)更新周期特別長、數(shù)據(jù)準(zhǔn)確的要求非常高等，所以我們也只是在部分場(chǎng)景進(jìn)行了嘗試性的探索。

對(duì)于單表計(jì)算場(chǎng)景，我們用canal實(shí)時(shí)讀取MySQL的BinLog，經(jīng)由Kafka輸入到Flink中進(jìn)行計(jì)算，最后寫入HBase的結(jié)果表中。這個(gè)體驗(yàn)還是非常舒服的。

對(duì)于多表關(guān)聯(lián)計(jì)算場(chǎng)景，我們先后嘗試了4種解決方案，各有利弊。

最早的方案是所有需要關(guān)聯(lián)的表都在Flink中做實(shí)時(shí)關(guān)聯(lián)，然后存到HBase中。當(dāng)時(shí)用的是Flink1.5，不支持MySQL當(dāng)維表，所以還用Flink實(shí)時(shí)同步數(shù)據(jù)到HBase，做成實(shí)時(shí)維度。這個(gè)方案看上去簡單，但是維表數(shù)據(jù)經(jīng)常延遲，關(guān)聯(lián)不上。

然后我們把Flink更新到1.10，直接用MySQL的表當(dāng)維表，比早期的方案更簡化了，維表也沒有延遲的問題。但是隨之發(fā)現(xiàn)維表更新的時(shí)候，不能觸發(fā)計(jì)算。這是因?yàn)镕link是事件驅(qū)動(dòng)，只有主表更新時(shí)才會(huì)觸發(fā)，維表更新的時(shí)候就不動(dòng)了，這個(gè)比較惱火。

因?yàn)檫@個(gè)缺陷嚴(yán)重影響業(yè)務(wù)，所以我們又做了一次嘗試，把所有表都放到Kafka中，然后再到Flink中進(jìn)行多表關(guān)聯(lián)。這樣每個(gè)表更新都可以觸發(fā)計(jì)算。這樣基本上需求就都能滿足了。

但是上面的方案依然有缺點(diǎn)，結(jié)果表每次都會(huì)讀取全面字段后再進(jìn)行更新，效率不夠高。

于是我們?cè)俅握{(diào)整方案，依然是把結(jié)果表當(dāng)維表關(guān)聯(lián)，但是每個(gè)任務(wù)只更新自己相關(guān)的字段，這樣就規(guī)避了前面說的效率問題。

整個(gè)過程看下來，F(xiàn)link能解決咱的業(yè)務(wù)場(chǎng)景的問題，實(shí)時(shí)性也非常好。但是一旦到了某個(gè)具體的場(chǎng)景中，就會(huì)有無數(shù)問題等著你，解決一個(gè)，又來一個(gè)，考驗(yàn)的不僅是你對(duì)Flink的熟悉程度，還有遇到問題解決問題的能力，更考驗(yàn)架構(gòu)的能力。

這不正如我們的人生么？總是有一個(gè)有一個(gè)的挑戰(zhàn)等著我們。只要你持續(xù)精進(jìn)，終歸是能找到解決辦法的。加油！

我是大數(shù)據(jù)架構(gòu)師鳴宇淳，歡迎你跟我一起聊大數(shù)據(jù)，我的微信號(hào)：chybin001。

擴(kuò)展閱讀：1本Flink電子書+9份分享ppt，公眾號(hào)“大數(shù)據(jù)架構(gòu)師”后臺(tái)回復(fù)“Flink”即可下載。

配合以下文章享受更佳