Warehouse Syncs API

Προγραμματισμένες εισαγωγές από την αποθήκη δεδομένων σας - BigQuery, Snowflake ή Redshift - στις ίδιες διαδρομές εγγραφής που χρησιμοποιεί μια εισαγωγή αρχείου.

Όλα τα endpoints αυτής της σελίδας είναι σχετικά με τη βασική διεύθυνση: https://api-eu1.joryio.com/api - δείτε Επισκόπηση API.

Επισκόπηση

Ένας συγχρονισμός αποθήκης διαβάζει έναν πίνακα ή view που ετοιμάζετε εσείς και γράφει κάθε σελίδα γραμμών στο Joryio καθώς φτάνει. Τίποτα δεν συσσωρεύεται στη μνήμη, οπότε ένας συγχρονισμός δέκα εκατομμυρίων γραμμών κοστίζει σε μνήμη όσο και ένας των δέκα χιλιάδων.

Δύο πράγματα που ένας συγχρονισμός δεν είναι:

• Δεν είναι ελεύθερη SQL. Το ερώτημα παράγεται σε κάθε εκτέλεση ώστε να φέρει τη σταδιακή συνθήκη, γι' αυτό δίνετε όνομα πίνακα ή view και μια στήλη watermark, όχι ένα SELECT.
• Δεν είναι ξεχωριστή διαδρομή εισαγωγής. Οι γραμμές γράφονται μέσα από τον ίδιο ακριβώς κώδικα που χρησιμοποιεί μια εισαγωγή αρχείου, μαζί με τους κανόνες μετάβασης της συγκατάθεσης - έτσι μια απεγγραφή στην αποθήκη σας τηρείται, ενώ μια επανεγγραφή δεν γίνεται, εκτός αν το ζητήσετε ρητά.

Έλεγχος ταυτότητας

Authorization: Bearer YOUR_API_KEY

Η ανάγνωση ενός συγχρονισμού απαιτεί integrations:read· η δημιουργία, ενημέρωση, εκτέλεση ή διαγραφή απαιτεί integrations:write. Η ενεργοποίηση του overrideConsent απαιτεί επιπλέον compliance:write.

Το αντικείμενο συγχρονισμού

Πεδίο	Τύπος	Σημειώσεις
integrationId	uuid	Η αποθηκευμένη σύνδεση αποθήκης μέσω της οποίας διαβάζουμε
connectorId	string	bigquery, snowflake ή redshift
syncType	string	users, events ή deletions - καθορίζει σε ποια διαδρομή εγγραφής πάνε οι γραμμές
sourceTable	string	table, schema.table ή db.schema.table
watermarkColumn	string?	Η στήλη που δείχνει ότι μια γραμμή άλλαξε. Παράλειψη = πλήρης ανανέωση σε κάθε εκτέλεση
keyColumn	string?	Μοναδική, ταξινομήσιμη στήλη για ισοπαλίες. Συνιστάται
watermark	timestamp?	Μέχρι πού έχει διαβάσει ο συγχρονισμός. Το διαχειριζόμαστε εμείς· καθαρίστε το με Rewind
columnMapping	object	Στήλη αποθήκης προς διαδρομή πεδίου προορισμού
schedule	string	manual, every_5_min, every_15_min, hourly, every_6_hours, daily, weekly, monthly
cadence	object	Μόνο για ανάγνωση. configuredMs, effectiveMs, throttled - δείτε παρακάτω
enabled	boolean	Απενεργοποιημένο σημαίνει ότι ο χρονοπρογραμματιστής το παρακάμπτει, διατηρώντας τη συχνότητα
overrideConsent	boolean	Η αποθήκη ως πηγή αλήθειας για τη συγκατάθεση. Απαιτεί compliance:write
lastRunStats	object?	Γραμμές που διαβάστηκαν, δημιουργήθηκαν, ενημερώθηκαν, παραλείφθηκαν, απέτυχαν, διάρκεια και δείγμα σφαλμάτων

Αντιστοίχιση στηλών

Προορισμός	Σημασία
email, phone, externalId	Ταυτότητα. Απαιτείται τουλάχιστον ένα από email ή externalId ανά γραμμή
attributes.<key>	Ιδιότητα προφίλ. Το κλειδί δεν μπορεί να περιέχει τελεία
subscriptions.<channel>	Καθολική συγκατάθεση καναλιού - email, sms, whatsapp, push, viber
subscriptions.list.<listId>.<channel>	Συγκατάθεση σε ΜΙΑ ομάδα συνδρομής
userId, eventName, timestamp, properties.<key>	Για syncType: events

Διαγραφές

Ένας συγχρονισμός deletions διαβάζει έναν πίνακα με άτομα προς διαγραφή. Δεν διαγράφει ο ίδιος - δημιουργεί αιτήματα διαγραφής στην κανονική διαδρομή συμμόρφωσης, την ίδια που χρησιμοποιεί το DSR API, με το πλήρες audit trail.

• Μια γραμμή που ονομάζει κάποιον που δεν εισήχθη ποτέ, ή έχει ήδη διαγραφεί, παραλείπεται, δεν αποτυγχάνει.
• Κάποιος που ήδη διαγράφεται δεν μπαίνει ξανά σε ουρά.

Συγκατάθεση

Η συγκατάθεση κινείται μέσω ενός συγχρονισμού με τους ίδιους κανόνες όπως σε μια εισαγωγή αρχείου:

• Μια απεγγραφή τηρείται πάντα, χωρίς δικαίωμα και χωρίς σημαία. Ο συγχρονισμός απεγγραφών από το CRM σας δεν χρειάζεται καμία ενεργοποίηση.
• Ένας συγχρονισμός δεν θα επανεγγράψει κάποιον που απεγγράφηκε και δεν θα υποβαθμίσει μια επιβεβαιωμένη διπλή συγκατάθεση, εκτός αν οριστεί overrideConsent.

Το overrideConsent σημαίνει περισσότερα σε έναν συγχρονισμό απ' ό,τι σε ένα αρχείο. Ένα αρχείο εφαρμόζεται μία φορά· ένας συγχρονισμός επαναλαμβάνει την πηγή του σε κάθε κύκλο, οπότε μια παράκαμψη που έμεινε ενεργή αναιρεί κάθε απεγγραφή που φτάνει ανάμεσα στις εκτελέσεις, για όσο υπάρχει ο συγχρονισμός.

Endpoints

Λίστα συγχρονισμών

GET /warehouse-syncs

Ανάκτηση συγχρονισμού

GET /warehouse-syncs/:id

Λίστα εκτελέσεων

GET /warehouse-syncs/:id/runs?limit=50

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

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

Δημιουργία συγχρονισμού

POST /warehouse-syncs

{
  "integrationId": "0f8e...",
  "connectorId": "snowflake",
  "name": "Contacts sync",
  "syncType": "users",
  "sourceTable": "analytics.public.contacts",
  "watermarkColumn": "updated_at",
  "schedule": "hourly",
  "columnMapping": {
    "EMAIL": "email",
    "CUSTOMER_ID": "externalId",
    "PLAN": "attributes.plan",
    "EMAIL_CONSENT": "subscriptions.email"
  }
}

Τα sourceTable και watermarkColumn πρέπει να είναι απλά αναγνωριστικά. Οτιδήποτε άλλο - εισαγωγικά, κενά, ερωτηματικά, σύμβολα σχολίων - απορρίπτεται με 400, επειδή αυτά τα ονόματα ενσωματώνονται σε SQL που παράγεται αυτόματα.

Ενημέρωση συγχρονισμού

PATCH /warehouse-syncs/:id

Η δημιουργία και η ενημέρωση δέχονται μόνο τα πεδία που μπορείτε να ορίσετε: name, integrationId, syncType, sourceTable, sourceQuery, watermarkColumn, keyColumn, columnMapping, schedule, enabled, overrideConsent. Οποιοδήποτε άλλο πεδίο - id, organizationId, workspaceId, status, watermark ή οι μετρητές εκτελέσεων - απορρίπτεται με 400· αυτά τα διαχειρίζεται το Joryio. Το integrationId πρέπει να είναι σύνδεση αποθήκης δεδομένων στον ίδιο χώρο εργασίας με τον συγχρονισμό, αλλιώς το αίτημα επιστρέφει 404. Το connectorId είναι προαιρετικό και λαμβάνεται από τη σύνδεση - αν το στείλετε, πρέπει να ταιριάζει.

Άμεση εκτέλεση

POST /warehouse-syncs/:id/run

Επιστρέφει queued: false όταν ο συγχρονισμός είναι ήδη σε ουρά ή εκτελείται. Ένας συγχρονισμός που επικαλύπτει τον εαυτό του διαβάζει τις ίδιες γραμμές δύο φορές και διπλασιάζει τον λογαριασμό της αποθήκης χωρίς νέα δεδομένα.

Rewind (μηδενισμός watermark)

POST /warehouse-syncs/:id/rewind

Καθαρίζει το watermark ώστε η επόμενη εκτέλεση να διαβάσει τον πίνακα από την αρχή. Είναι ασφαλές - κάθε εγγραφή είναι ένα ιδεμποτικό upsert με βάση την ταυτότητα - και είναι η διόρθωση όταν μια αντιστοίχιση στηλών ήταν λάθος: διορθώστε την, κάντε rewind, εκτελέστε.

Διαγραφή

DELETE /warehouse-syncs/:id

Πραγματική συχνότητα

Μια εκτέλεση που διαρκεί περισσότερο από το διάστημά της απορρίπτεται, δεν μπαίνει σε ουρά - ένας συγχρονισμός δεν επικαλύπτει ποτέ τον εαυτό του. Έτσι ένα πρόγραμμα 5 λεπτών σε συγχρονισμό που διαρκεί 7 λεπτά εκτελείται στην πραγματικότητα κάθε 7 λεπτά, και το cadence.throttled σας το λέει.

Το every_5_min είναι το πρακτικό κατώφλι για μια προγραμματισμένη ανάγνωση, και ο λόγος είναι το κόστος: το Snowflake χρεώνει ανά δευτερόλεπτο λειτουργίας με ελάχιστο 60 δευτερολέπτων σε κάθε επανεκκίνηση, οπότε ένας τόσο συχνός συγχρονισμός δεν αφήνει ποτέ την αποθήκη να κοιμηθεί.

Μηνιαίως

Το monthly διατηρεί την ημέρα του μήνα από την οποία ξεκίνησε ο συγχρονισμός. Ένας συγχρονισμός αγκυρωμένος στις 31 εκτελείται στις 28 Φεβρουαρίου (29 σε δίσεκτο έτος) και επιστρέφει στις 31 Μαρτίου - δεν μεταναστεύει στις 28 για πάντα. Όλη η αριθμητική των χρονοπρογραμμάτων γίνεται σε UTC, επειδή ένας συγχρονισμός δεν φέρει ζώνη ώρας και η τοπική αριθμητική θα σήμαινε σιωπηλά τη ζώνη του μηχανήματος που τυχαίνει να τρέχει τον χρονοπρογραμματιστή.

Πώς λειτουργούν οι σταδιακές αναγνώσεις

Με ορισμένη watermarkColumn, κάθε εκτέλεση διαβάζει γραμμές από το αποθηκευμένο watermark και μετά, ταξινομημένες κατά αυτή τη στήλη. Το watermark προχωράει μόνο αφού μια σελίδα γραφτεί με ανθεκτικό τρόπο, ώστε μια εκτέλεση που διακόπηκε να ξαναδιαβάζει την ουρά της αντί να την προσπερνά.

Με μια keyColumn, το πρόβλημα του ορίου εξαφανίζεται αντί να το διαχειριζόμαστε. Ο δρομέας γίνεται (watermarkColumn, keyColumn) και κάθε εκτέλεση διαβάζει αυστηρά μετά την τελευταία γραμμή που διάβασε:

WHERE updated_at > :ts OR (updated_at = :ts AND id > :lastId)
ORDER BY updated_at, id

Οι γραμμές με την ίδια χρονοσφραγίδα διαβάζονται ακριβώς μία φορά - ούτε χάνονται ούτε ξαναδιαβάζονται - και δεν χρειάζεται επιπλέον ερώτημα καταμέτρησης. Επιλέξτε κάτι μονότονο: ένα αυτόματα αυξανόμενο id ή ένα ULID. Ένα hash περιεχομένου είναι σταθερό αλλά τυχαία ταξινομημένο, οπότε μια γραμμή που εισάγεται αργότερα την ίδια στιγμή θα ταξινομούνταν πριν από τον δρομέα και δεν θα διαβαζόταν ποτέ.

Χωρίς keyColumn επιστρέφουμε σε δρομέα μόνο χρονοσφραγίδας, και το όριο αντιμετωπίζεται με έναν από δύο τρόπους:

• Αποτυπώματα (προεπιλογή, έως 5.000 γραμμές στο όριο). Θυμόμαστε ποιες γραμμές βρίσκονταν στη στιγμή του ορίου και παραλείπουμε ακριβώς αυτές στην επόμενη εκτέλεση. Το όριο ξαναδιαβάζεται κάθε φορά ώστε να αναγνωρίζονται, και τίποτα δεν χάνεται.
• Καταμετρήσεις γραμμών (πάνω από 5.000). Θυμόμαστε πόσες γραμμές βρίσκονταν εκεί και ξαναδιαβάζουμε μόνο όταν ο αριθμός αυξηθεί. Φθηνότερο, επειδή το όριο δεν ξαναδιαβάζεται, και έχει ένα κενό: μια γραμμή που διαγράφεται και μια που εισάγεται την ίδια στιγμή αφήνουν την καταμέτρηση αμετάβλητη, οπότε η νέα γραμμή δεν διαβάζεται ποτέ.

Μια keyColumn αποφεύγει και τους δύο συμβιβασμούς - οι γραμμές του ορίου εξαιρούνται στον όρο WHERE, οπότε ούτε ξαναδιαβάζονται ούτε χάνονται. Είναι η συνιστώμενη ρύθμιση ακριβώς γι' αυτόν τον λόγο.

Το όριο είναι συμπεριληπτικό (>=) σκόπιμα. Ένα αποκλειστικό όριο χάνει σιωπηλά γραμμές που μοιράζονται ακριβώς τη χρονοσήμανση του watermark αλλά δεν είχαν διαβαστεί όταν τελείωσε η εκτέλεση. Ένα συμπεριληπτικό ξαναδιαβάζει λίγες γραμμές, κάτι που δεν κοστίζει τίποτα: κάθε εγγραφή είναι ιδεμποτικό upsert, οπότε μια γραμμή που φτάνει δύο φορές αποθηκεύεται μία.

Προεπισκόπηση γραμμών προέλευσης

POST /warehouse-syncs/preview

{ "integrationId": "0f8e...", "sourceTable": "analytics.public.contacts", "limit": 20 }

Μια περιορισμένη ανάγνωση της πηγής όπως είναι τώρα - χωρίς watermark, χωρίς σταδιακό κατηγόρημα - ώστε να χτίσετε την αντιστοίχιση στηλών με τα πραγματικά ονόματα στηλών και όχι με όσα θυμάστε. Στείλτε ακριβώς ένα από τα sourceTable ή sourceQuery. Εκτελείται άμεσα και έχει αυστηρό όριο 100 γραμμών.