โหลดข้อมูลจากคลังข้อมูลไปยังเลคเฮ้าส์แบบเพิ่มหน่วย

ในบทช่วยสอนนี้ คุณจะได้เรียนรู้วิธีการโหลดข้อมูลจาก Data Warehouse ไปยัง Lakehouse แบบเพิ่มหน่วย

ภาพรวม

นี่คือไดอะแกรมโซลูชันระดับสูง:

นี่คือขั้นตอนสําคัญในการสร้างโซลูชันนี้:

1. เลือกคอลัมน์ลายน้ํา เลือกหนึ่งคอลัมน์ในตารางข้อมูลต้นทาง ซึ่งสามารถใช้เพื่อแบ่งส่วนระเบียนใหม่หรือระเบียนที่อัปเดตแล้วสําหรับทุกการเรียกใช้ โดยปกติข้อมูลในคอลัมน์ที่เลือกนี้ (ตัวอย่างเช่น last_modify_time หรือ ID) ยังคงเพิ่มขึ้นเมื่อมีการสร้างหรืออัปเดตแถว ค่าสูงสุดในคอลัมน์นี้จะใช้เป็นลายน้ํา

2. เตรียมตารางเพื่อจัดเก็บค่าลายน้ําล่าสุดในคลังข้อมูลของคุณ

3. สร้างไปป์ไลน์ด้วยเวิร์กโฟลว์ต่อไปนี้:

   ไปป์ไลน์ในโซลูชันนี้มีกิจกรรมต่อไปนี้:

   • สร้างกิจกรรมการค้นหาสองรายการ ใช้กิจกรรมการค้นหาแรกเพื่อดึงค่าลายน้ําล่าสุด ใช้กิจกรรมการค้นหาที่สองเพื่อดึงค่าลายน้ําใหม่ ค่าลายน้ําเหล่านี้จะถูกส่งผ่านไปยังกิจกรรมการคัดลอก
   • สร้างกิจกรรมการคัดลอกที่คัดลอกแถวจากตารางข้อมูลต้นทางที่มีค่าของคอลัมน์ลายน้ํามากกว่าค่าลายน้ําเก่าและน้อยกว่าค่าลายน้ําใหม่ จากนั้นคัดลอกข้อมูลจากคลังข้อมูลไปยังเลคเฮ้าส์เป็นไฟล์ใหม่
   • สร้างกิจกรรมของ Stored Procedure ที่อัปเดตค่าลายน้ําล่าสุดสําหรับการเรียกใช้ไปป์ไลน์ถัดไป

ข้อกำหนดเบื้องต้น

• คลังข้อมูล คุณใช้ Data Warehouse เป็นที่เก็บข้อมูลต้นทาง ถ้าคุณยังไม่มี ให้ดู สร้างคลังข้อมูล สําหรับขั้นตอนในการสร้างคลัง ข้อมูล
• เลคเฮ้าส์ คุณใช้เลคเฮ้าส์เป็นที่เก็บข้อมูลปลายทาง ถ้าคุณยังไม่มี ดู สร้างเลคเฮ้าส์ สําหรับขั้นตอนในการสร้าง สร้างโฟลเดอร์ที่ชื่อว่า IncrementalCopy เพื่อจัดเก็บข้อมูลที่คัดลอก

การเตรียมแหล่งข้อมูลของคุณ

นี่คือตารางบางตารางและกระบวนงานที่เก็บไว้ที่คุณต้องเตรียมในคลังข้อมูลต้นทางของคุณก่อนที่จะกําหนดค่าไปป์ไลน์การคัดลอกแบบเพิ่มหน่วย

1. สร้างตารางแหล่งข้อมูลในคลังข้อมูลของคุณ

เรียกใช้คําสั่ง SQL ต่อไปนี้ในคลังข้อมูลของคุณเพื่อสร้างตารางที่ ชื่อ data_source_table เป็นตารางแหล่งข้อมูล ในบทช่วยสอนนี้ คุณจะใช้สิ่งนี้เป็นข้อมูลตัวอย่างเพื่อทําสําเนาแบบเพิ่มหน่วย

create table data_source_table
(
    PersonID int,
    Name varchar(255),
    LastModifytime DATETIME2(6)
);

INSERT INTO data_source_table
    (PersonID, Name, LastModifytime)
VALUES
    (1, 'aaaa','9/1/2017 12:56:00 AM'),
    (2, 'bbbb','9/2/2017 5:23:00 AM'),
    (3, 'cccc','9/3/2017 2:36:00 AM'),
    (4, 'dddd','9/4/2017 3:21:00 AM'),
    (5, 'eeee','9/5/2017 8:06:00 AM');

ข้อมูลในตารางแหล่งข้อมูลจะแสดงอยู่ด้านล่าง:

PersonID | Name | LastModifytime
-------- | ---- | --------------
1        | aaaa | 2017-09-01 00:56:00.000
2        | bbbb | 2017-09-02 05:23:00.000
3        | cccc | 2017-09-03 02:36:00.000
4        | dddd | 2017-09-04 03:21:00.000
5        | eeee | 2017-09-05 08:06:00.000

ในบทช่วยสอนนี้ คุณใช้ LastModifytime เป็นคอลัมน์ลายน้ํา

2. สร้างตารางอื่นในคลังข้อมูลเพื่อจัดเก็บค่าลายน้ําล่าสุด

1. เรียกใช้คําสั่ง SQL ต่อไปนี้ในคลังข้อมูลของคุณเพื่อสร้างตารางที่ ชื่อว่า watermarktable เพื่อจัดเก็บค่าลายน้ําล่าสุด:

   create table watermarktable
   (
   TableName varchar(255),
   WatermarkValue DATETIME2(6),
   );
   

2. ตั้งค่าค่าเริ่มต้นของลายน้ําสุดท้ายด้วยชื่อตารางของตารางข้อมูลต้นทาง ในบทช่วยสอนนี้ ชื่อตารางจะถูก data_source_table และค่าเริ่มต้นคือ1/1/2010 12:00:00 AM

   INSERT INTO watermarktable
   VALUES ('data_source_table','1/1/2010 12:00:00 AM')    
   

3. ตรวจทานข้อมูลในแถบน้ําตาราง

   Select * from watermarktable
   

   เอาท์พุท

   TableName  | WatermarkValue
   ----------  | --------------
   data_source_table | 2010-01-01 00:00:00.000
   

3. สร้างขั้นตอนที่จัดเก็บไว้ในคลังข้อมูลของคุณ

เรียกใช้คําสั่งต่อไปนี้เพื่อสร้างกระบวนงานที่จัดเก็บไว้ในคลังข้อมูลของคุณ กระบวนการที่เก็บไว้นี้ใช้เพื่อช่วยอัปเดตค่าลายน้ําล่าสุดหลังจากเรียกใช้ไปป์ไลน์ล่าสุด

CREATE PROCEDURE usp_write_watermark @LastModifiedtime datetime, @TableName varchar(50)
AS

BEGIN

UPDATE watermarktable
SET [WatermarkValue] = @LastModifiedtime
WHERE [TableName] = @TableName

END

กําหนดค่าไปป์ไลน์สําหรับสําเนาแบบเพิ่มหน่วย

ขั้นตอนที่ 1: สร้างไปป์ไลน์

1. นําทางไปยัง Power BI

2. เลือกไอคอน Power BI ที่ด้านล่างซ้ายของหน้าจอ จากนั้นเลือก ข้อมูลจากโรงงาน เพื่อเปิดหน้าแรกของ Data Factory

3. นําทางไปยังพื้นที่ทํางาน Microsoft Fabric ของคุณ

4. เลือก ไปป์ไลน์ ข้อมูล จากนั้นป้อนชื่อไปป์ไลน์เพื่อสร้างไปป์ไลน์ใหม่

   

   

ขั้นตอนที่ 2: เพิ่มกิจกรรมการค้นหาสําหรับลายน้ําล่าสุด

ในขั้นตอนนี้ คุณสร้างกิจกรรมการค้นหาเพื่อรับค่าลายน้ําล่าสุด ค่าเริ่มต้น 1/1/2010 12:00:00 AM ตั้งค่าก่อนหน้าจะได้รับ

1. เลือก เพิ่มกิจกรรม ไปป์ไลน์ และเลือก ค้นหา จากรายการดรอปดาวน์

2. ภายใต้แท็บ ทั่วไป เปลี่ยนชื่อกิจกรรมนี้เป็น LookupOldWaterMarkActivity

3. ภายใต้ แท็บ การตั้งค่า ดําเนินการกําหนดค่าต่อไปนี้:

   • ชนิดของที่เก็บข้อมูล: เลือกพื้นที่ทํางาน
   • ประเภทที่เก็บข้อมูลของพื้นที่ทํางาน: เลือก คลังข้อมูล
   • คลังข้อมูล: เลือกคลังข้อมูลของคุณ
   • ใช้คิวรี: เลือกตาราง
   • ตาราง: เลือก dbo.watermarktable
   • แถวแรกเท่านั้น: เลือกแล้ว

   

ขั้นตอนที่ 3: เพิ่มกิจกรรมการค้นหาสําหรับลายน้ําใหม่

ในขั้นตอนนี้ คุณสร้างกิจกรรมการค้นหาเพื่อรับค่าลายน้ําใหม่ คุณใช้คิวรีเพื่อขอรับลายน้ําใหม่จากตารางข้อมูลต้นทางของคุณ ได้รับค่าสูงสุดในคอลัมน์ LastModifytime ใน data_source_table

1. บนแถบด้านบนสุด เลือก ค้นหา ภายใต้ แท็บกิจกรรม เพื่อเพิ่มกิจกรรมการค้นหาที่สอง

2. ภายใต้แท็บ ทั่วไป เปลี่ยนชื่อกิจกรรมนี้เป็น LookupNewWaterMarkActivity

3. ภายใต้ แท็บ การตั้งค่า ดําเนินการกําหนดค่าต่อไปนี้:

   • ชนิดของที่เก็บข้อมูล: เลือกพื้นที่ทํางาน

   • ประเภทที่เก็บข้อมูลของพื้นที่ทํางาน: เลือก คลังข้อมูล

   • คลังข้อมูล: เลือกคลังข้อมูลของคุณ

   • ใช้คิวรี: เลือกคิวรี

   • คิวรี: ใส่คิวรีต่อไปนี้เพื่อเลือกเวลาที่ปรับเปลี่ยนล่าสุดเป็นลายน้ําใหม่สูงสุด:

     select MAX(LastModifytime) as NewWatermarkvalue from data_source_table
     

   • แถวแรกเท่านั้น: เลือกแล้ว

   

ขั้นตอนที่ 4: เพิ่มกิจกรรมการคัดลอกเพื่อคัดลอกข้อมูลแบบเพิ่มหน่วย

ในขั้นตอนนี้ คุณเพิ่มกิจกรรมการคัดลอกเพื่อคัดลอกข้อมูลแบบเพิ่มหน่วยระหว่างลายน้ําล่าสุดและลายน้ําใหม่จาก Data Warehouse ไปยัง Lakehouse

1. เลือก กิจกรรม บนแถบด้านบนและเลือก คัดลอกข้อมูล ->เพิ่มไปยังพื้นที่ทํางาน เพื่อรับกิจกรรมการคัดลอก

2. ภายใต้แท็บ ทั่วไป เปลี่ยนชื่อกิจกรรมนี้เป็น IncrementalCopyActivity

3. เชื่อมต่อกิจกรรมการค้นหาทั้งสองไปยังกิจกรรมการคัดลอกโดยการลากปุ่มสีเขียว (บนความสําเร็จ) ที่แนบไปกับกิจกรรมการค้นหาไปยังกิจกรรมการคัดลอก ปล่อยปุ่มเมาส์เมื่อคุณเห็นสีเส้นขอบของกิจกรรมการคัดลอกเปลี่ยนเป็นสีเขียว

   

4. ภายใต้ แท็บ แหล่งข้อมูล ทําการกําหนดค่าต่อไปนี้:

   • ชนิดของที่เก็บข้อมูล: เลือกพื้นที่ทํางาน

   • ประเภทที่เก็บข้อมูลของพื้นที่ทํางาน: เลือก คลังข้อมูล

   • คลังข้อมูล: เลือกคลังข้อมูลของคุณ

   • ใช้คิวรี: เลือกคิวรี

   • คิวรี: ใส่คิวรีต่อไปนี้เพื่อคัดลอกข้อมูลแบบเพิ่มหน่วยระหว่างลายน้ําล่าสุดและลายน้ําใหม่

     select * from data_source_table where LastModifytime > '@{activity('LookupOldWaterMarkActivity').output.firstRow.WatermarkValue}' and LastModifytime <= '@{activity('LookupNewWaterMarkActivity').output.firstRow.NewWatermarkvalue}'
     

   

5. ภายใต้ แท็บปลายทาง ดําเนินการกําหนดค่าต่อไปนี้:

   • ชนิดของที่เก็บข้อมูล: เลือกพื้นที่ทํางาน
   • ประเภทที่เก็บข้อมูลพื้นที่ทํางาน: เลือกเลคเฮ้าส์
   • เลคเฮ้าส์: เลือกเลคเฮาส์ของคุณ
   • โฟลเดอร์ราก: เลือกไฟล์
   • พาธของไฟล์: ระบุโฟลเดอร์ที่คุณต้องการจัดเก็บข้อมูลที่คัดลอกของคุณ เลือก เรียกดู เพื่อเลือกโฟลเดอร์ของคุณ สําหรับชื่อไฟล์ ให้เปิด เพิ่มเนื้อหา แบบไดนามิก และป้อน @CONCAT('Incremental-', pipeline().RunId, '.txt') ในหน้าต่างที่เปิดอยู่เพื่อสร้างชื่อไฟล์สําหรับไฟล์ข้อมูลที่คัดลอกของคุณใน Lakehouse
   • รูปแบบไฟล์: เลือกประเภทรูปแบบของข้อมูลของคุณ

   

ขั้นตอนที่ 5: เพิ่มกิจกรรมของกระบวนงานที่เก็บไว้

ในขั้นตอนนี้ คุณเพิ่มกิจกรรมของ Stored Procedure เพื่ออัปเดตค่าลายน้ําล่าสุดสําหรับการเรียกใช้ไปป์ไลน์ถัดไป

1. เลือก กิจกรรม บนแถบด้านบนและเลือก Stored Procedure เพื่อเพิ่มกิจกรรมของ Stored Procedure

2. ภายใต้แท็บ ทั่วไป เปลี่ยนชื่อกิจกรรมนี้เป็น StoredProceduretoWriteWatermarkActivity

3. เชื่อมต่อเอาต์พุตสีเขียว (เมื่อประสบความสําเร็จ) ของกิจกรรมการคัดลอกไปยังกิจกรรมของ Stored Procedure

4. ภายใต้ แท็บ การตั้งค่า ดําเนินการกําหนดค่าต่อไปนี้:

   • ชนิดของที่เก็บข้อมูล: เลือกพื้นที่ทํางาน

   • คลังข้อมูล: เลือกคลังข้อมูลของคุณ

   • ชื่อ Stored Procedure: ระบุกระบวนงานที่เก็บไว้ที่คุณสร้างขึ้นในคลังข้อมูลของคุณ: [dbo][ usp_write_watermark]

   • ขยาย พารามิเตอร์ Stored Procedure หากต้องการระบุค่าสําหรับพารามิเตอร์ Stored Procedure ให้เลือก นําเข้า แล้วป้อนค่าต่อไปนี้สําหรับพารามิเตอร์:

     ชื่อ	ชนิด	ค่า
     LastModifiedtime	Datetime	@{activity('LookupNewWaterMarkActivity').output.firstRow.NewWatermarkvalue}
     TableName	สตริง	@{activity('LookupOldWaterMarkActivity').output.firstRow.TableName}

   

ขั้นตอนที่ 6: เรียกใช้ไปป์ไลน์และตรวจสอบผลลัพธ์

บนแถบด้านบนสุด เลือก เรียกใช้ ภายใต้แท็บ หน้าแรก จากนั้นเลือก บันทึกและเรียกใช้ ไปป์ไลน์เริ่มต้นทํางานและคุณสามารถตรวจสอบไปป์ไลน์ภายใต้แท็บเอาต์พุต

ไปยังเลคเฮ้าส์ของคุณ คุณพบว่าไฟล์ข้อมูลอยู่ภายใต้โฟลเดอร์ที่คุณระบุ และคุณสามารถเลือกไฟล์เพื่อแสดงตัวอย่างข้อมูลที่คัดลอก

เพิ่มข้อมูลเพิ่มเติมเพื่อดูผลลัพธ์การคัดลอกแบบเพิ่มหน่วย

หลังจากที่คุณดําเนินการเรียกใช้ไปป์ไลน์แรกเสร็จสิ้นแล้ว ลองเพิ่มข้อมูลเพิ่มเติมในตารางแหล่งข้อมูล Data Warehouse ของคุณเพื่อดูว่าไปป์ไลน์นี้สามารถคัดลอกข้อมูลแบบเพิ่มหน่วยของคุณได้หรือไม่

ขั้นตอนที่ 1: เพิ่มข้อมูลเพิ่มเติมไปยังแหล่งข้อมูล

แทรกข้อมูลใหม่ลงในคลังข้อมูลของคุณโดยการเรียกใช้คิวรีต่อไปนี้:

INSERT INTO data_source_table
VALUES (6, 'newdata','9/6/2017 2:23:00 AM')

INSERT INTO data_source_table
VALUES (7, 'newdata','9/7/2017 9:01:00 AM')

ข้อมูลที่อัปเดตสําหรับ data_source_table คือ:

PersonID | Name | LastModifytime
-------- | ---- | --------------
1 | aaaa | 2017-09-01 00:56:00.000
2 | bbbb | 2017-09-02 05:23:00.000
3 | cccc | 2017-09-03 02:36:00.000
4 | dddd | 2017-09-04 03:21:00.000
5 | eeee | 2017-09-05 08:06:00.000
6 | newdata | 2017-09-06 02:23:00.000
7 | newdata | 2017-09-07 09:01:00.000

ขั้นตอนที่ 2: ทริกเกอร์การเรียกใช้ไปป์ไลน์อื่นและตรวจสอบผลลัพธ์

กลับไปยังหน้าไปป์ไลน์ของคุณ บนแถบด้านบน ให้เลือก เรียกใช้ ภายใต้ แท็บ หน้าแรก อีกครั้ง ไปป์ไลน์เริ่มต้นทํางานและคุณสามารถตรวจสอบไปป์ไลน์ภายใต้ผลลัพธ์

ไปยังเลคเฮ้าส์ของคุณ คุณจะพบไฟล์ข้อมูลที่คัดลอกใหม่อยู่ภายใต้โฟลเดอร์ที่คุณระบุ และคุณสามารถเลือกไฟล์เพื่อแสดงตัวอย่างข้อมูลที่คัดลอก คุณจะเห็นข้อมูลแบบเพิ่มหน่วยของคุณแสดงอยู่ในไฟล์นี้

เนื้อหาที่เกี่ยวข้อง

ถัดไป กรุณาเรียนรู้เพิ่มเติมเกี่ยวกับการคัดลอกจาก Azure Blob Storage ไปยัง Lakehouse

คัดลอกจาก Azure Blob Storage ไปยัง Lakehouse