Κοινή χρήση μέσω

Επαυξητική φόρτωση δεδομένων από την Αποθήκη δεδομένων στο Lakehouse

  • Άρθρο

Σε αυτή την εκμάθηση, θα μάθετε πώς να φορτώνετε δεδομένα επαυξητικής φόρτωσης από το Data Warehouse στο Lakehouse.

Επισκόπηση

Ακολουθεί το διάγραμμα λύσης υψηλού επιπέδου:

Διάγραμμα που εμφανίζει λογική δεδομένων επαυξητικής φόρτωσης.

Ακολουθούν τα σημαντικά βήματα για τη δημιουργία αυτής της λύσης:

  1. Επιλέξτε τη στήλη υδατογραφήματος. Επιλέξτε μία στήλη στον πίνακα δεδομένων προέλευσης, η οποία μπορεί να χρησιμοποιηθεί για ανάλυση των νέων ή ενημερωμένων εγγραφών για κάθε εκτέλεση. Κανονικά, τα δεδομένα σε αυτήν την επιλεγμένη στήλη (για παράδειγμα, last_modify_time ή αναγνωριστικό) συνεχίζουν να αυξάνονται όταν δημιουργούνται ή ενημερώνονται γραμμές. Η μέγιστη τιμή σε αυτήν τη στήλη χρησιμοποιείται ως υδατογράφημα.

  2. Προετοιμάστε έναν πίνακα για να αποθηκεύσετε την τελευταία τιμή υδατογραφήματος στην Αποθήκη δεδομένων σας.

  3. Δημιουργήστε μια διοχέτευση με την ακόλουθη ροή εργασιών:

    Η διοχέτευση σε αυτήν τη λύση έχει τις ακόλουθες δραστηριότητες:

    • Δημιουργήστε δύο δραστηριότητες αναζήτησης. Χρησιμοποιήστε την πρώτη δραστηριότητα αναζήτησης για να ανακτήσετε την τελευταία τιμή υδατογραφήματος. Χρησιμοποιήστε τη δεύτερη δραστηριότητα αναζήτησης για να ανακτήσετε τη νέα τιμή υδατογραφήματος. Αυτές οι τιμές υδατογραφήματος μεταβιβάζονται στη δραστηριότητα αντιγραφής.
    • Δημιουργήστε μια δραστηριότητα αντιγραφής που αντιγράφει γραμμές από τον πίνακα δεδομένων προέλευσης με την τιμή της στήλης υδατογραφήματος μεγαλύτερη από την παλιά τιμή υδατογραφήματος και μικρότερη από τη νέα τιμή υδατογραφήματος. Στη συνέχεια, αντιγράφει τα δεδομένα από την Αποθήκη δεδομένων στο Lakehouse ως νέο αρχείο.
    • Δημιουργήστε μια αποθηκευμένη δραστηριότητα διαδικασίας που ενημερώνει την τελευταία τιμή υδατογραφήματος για την επόμενη εκτέλεση διοχέτευσης.

Προαπαιτούμενα στοιχεία

  • Αποθήκη δεδομένων. Χρησιμοποιείτε την Αποθήκη δεδομένων ως χώρο αποθήκευσης δεδομένων προέλευσης. Εάν δεν το έχετε, ανατρέξτε στο θέμα Δημιουργία αποθήκης δεδομένων για τα βήματα δημιουργίας μίας.
  • Ο Λέικχαουζ. Χρησιμοποιείτε το Lakehouse ως χώρο αποθήκευσης δεδομένων προορισμού. Εάν δεν το έχετε, ανατρέξτε στο θέμα Δημιουργία lakehouse για βήματα για να δημιουργήσετε ένα. Δημιουργήστε έναν φάκελο με την ονομασία 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 στην Αποθήκη δεδομένων σας για να δημιουργήσετε έναν πίνακα με το όνομα υδατογράφημα για την αποθήκευση της τελευταίας τιμής υδατογραφήματος:

    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 για να ανοίξετε την αρχική σελίδα του 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: Προσθήκη της δραστηριότητας αντιγραφής για την αντιγραφή επαυξητικών δεδομένων

Σε αυτό το βήμα, προσθέτετε μια δραστηριότητα αντιγραφής για να αντιγράψετε τα επαυξητικά δεδομένα μεταξύ του τελευταίου υδατογραφήματος και του νέου υδατογραφήματος από την Αποθήκη δεδομένων στο 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. Στην καρτέλα Προορισμός , εκτελέστε τις ακόλουθες ρυθμίσεις παραμέτρων:

    • Τύπος χώρου αποθήκευσης δεδομένων: Επιλέξτε Χώρος εργασίας.
    • Τύπος χώρου αποθήκευσης δεδομένων χώρου εργασίας: Επιλέξτε Lakehouse.
    • Lakehouse: Επιλέξτε το lakehouse σας.
    • Ριζικός φάκελος: Επιλογή αρχείων.
    • Διαδρομή αρχείου: Καθορίστε τον φάκελο στον οποίο θέλετε να αποθηκεύσετε τα δεδομένα που αντιγράψατε. Επιλέξτε Αναζήτηση για να επιλέξετε τον φάκελο. Για το όνομα του αρχείου, ανοίξτε το στοιχείο Προσθήκη δυναμικού περιεχομένου και πληκτρολογήστε @CONCAT('Incremental-', pipeline().RunId, '.txt') στο παράθυρο που άνοιξε για να δημιουργήσετε ονόματα αρχείων για το αρχείο δεδομένων που αντιγράψατε στο Lakehouse.
    • Μορφή αρχείου: Επιλέξτε τον τύπο μορφής των δεδομένων σας.

    Στιγμιότυπο οθόνης που εμφανίζει την αντιγραφή ρύθμισης παραμέτρων προορισμού.

Βήμα 5: Προσθήκη δραστηριότητας αποθηκευμένης διαδικασίας

Σε αυτό το βήμα, προσθέτετε μια δραστηριότητα αποθηκευμένης διαδικασίας για να ενημερώσετε την τελευταία τιμή υδατογραφήματος για την επόμενη εκτέλεση διοχέτευσης.

  1. Επιλέξτε Δραστηριότητες στην επάνω γραμμή και επιλέξτε Αποθηκευμένη διαδικασία για να προσθέσετε μια δραστηριότητα αποθηκευμένης διαδικασίας.

  2. Στην καρτέλα Γενικά , μετονομάστε αυτήν τη δραστηριότητα σε StoredProceduretoWriteWatermarkActivity.

  3. Συνδέστε το πράσινο (On success) αποτέλεσμα της δραστηριότητας αντιγραφής στη δραστηριότητα αποθηκευμένης διαδικασίας.

  4. Στην καρτέλα Ρυθμίσεις , εκτελέστε τις ακόλουθες ρυθμίσεις παραμέτρων:

    • Τύπος χώρου αποθήκευσης δεδομένων: Επιλέξτε Χώρος εργασίας.

    • Αποθήκη δεδομένων: Επιλέξτε την Αποθήκη δεδομένων σας.

    • Αποθηκευμένο όνομα διαδικασίας: Καθορίστε την αποθηκευμένη διαδικασία που δημιουργήσατε στην Αποθήκη δεδομένων σας: [dbo].[ usp_write_watermark].

    • Αναπτύξτε τις Παραμέτρους αποθηκευμένης διαδικασίας. Για να καθορίσετε τιμές για τις παραμέτρους αποθηκευμένη διαδικασίας, επιλέξτε Εισαγωγή και εισαγάγετε τις ακόλουθες τιμές για τις παραμέτρους:

      Ονομασία Τύπος Τιμή
      LastModifiedtime DateTime @{activity('LookupNewWaterMarkActivity').output.firstRow.NewWatermarkvalue}
      TableName Συμβολοσειρά @{activity('LookupOldWaterMarkActivity').output.firstRow.TableName}

    Στιγμιότυπο οθόνης που εμφανίζει τη ρύθμιση παραμέτρων δραστηριότητας αποθηκευμένη διαδικασίας.

Βήμα 6: Εκτέλεση της διοχέτευσης και παρακολούθηση του αποτελέσματος

Στην επάνω γραμμή, επιλέξτε Εκτέλεσηστην Αρχική καρτέλα. Στη συνέχεια, επιλέξτε Αποθήκευση και εκτέλεση. Η διοχέτευση ξεκινά την εκτέλεση και μπορείτε να παρακολουθήσετε τη διοχέτευση στην καρτέλα Έξοδος .

Στιγμιότυπο οθόνης που εμφανίζει τα αποτελέσματα εκτέλεσης διοχέτευσης.

Μεταβείτε στο Lakehouse σας, θα βρείτε ότι το αρχείο δεδομένων βρίσκεται κάτω από τον φάκελο που καθορίσατε και μπορείτε να επιλέξετε το αρχείο για προεπισκόπηση των αντιγραμμένων δεδομένων.

Στιγμιότυπο οθόνης που εμφανίζει δεδομένα lakehouse για την πρώτη εκτέλεση διοχέτευσης.

Στιγμιότυπο οθόνης που εμφανίζει την προεπισκόπηση δεδομένων lakehouse για την πρώτη εκτέλεση διοχέτευσης.

Προσθήκη περισσότερων δεδομένων για να δείτε τα αποτελέσματα επαυξητικής αντιγραφής

Αφού ολοκληρώσετε την πρώτη εκτέλεση διοχέτευσης, ας δοκιμάσουμε να προσθέσουμε περισσότερα δεδομένα στον πίνακα προέλευσης της Αποθήκης δεδομένων για να δούμε εάν αυτή η διοχέτευση μπορεί να αντιγράψει τα επαυξητικά δεδομένα σας.

Βήμα 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: Ενεργοποίηση άλλης εκτέλεσης διοχέτευσης και παρακολούθηση του αποτελέσματος

Επιστρέψτε στη σελίδα διοχέτευσης. Στην επάνω γραμμή, επιλέξτε Εκτέλεσηστην Αρχική καρτέλα ξανά. Η διοχέτευση ξεκινά την εκτέλεση και μπορείτε να παρακολουθήσετε τη διοχέτευση στην περιοχή Έξοδος.

Μεταβείτε στο Lakehouse σας, θα βρείτε ότι το νέο αρχείο αντιγραμμένων δεδομένων βρίσκεται κάτω από τον φάκελο που καθορίσατε και μπορείτε να επιλέξετε το αρχείο για προεπισκόπηση των αντιγραμμένων δεδομένων. Βλέπετε ότι τα επαυξητικά δεδομένα εμφανίζονται σε αυτό το αρχείο.

Στιγμιότυπο οθόνης που εμφανίζει δεδομένα lakehouse για τη δεύτερη εκτέλεση διοχέτευσης.

Στιγμιότυπο οθόνης που εμφανίζει την προεπισκόπηση δεδομένων lakehouse για τη δεύτερη εκτέλεση διοχέτευσης.

Σχετικό περιεχόμενο

Στη συνέχεια, προχωρήστε για να μάθετε περισσότερα σχετικά με την αντιγραφή από το Azure Blob Storage στο Lakehouse.

Αντιγραφή από το Azure Blob Storage στο Lakehouse