TRANSFORMS (SMT)

1. GDPR

GdprEmail

GdprPesel

GdprPhone

GdprPassportId

GdprCreditCard

2. FPE

Example:

{
"transforms":"fpe",
"transforms.fpe.type":"FpeField$Value",
"transforms.fpe.key":"3677397A24432646294A404D63516654",
"transforms.fpe.fields":"SENDER"
}

3. INSERT FIELD

Example:

{"transforms": "InsertField",
"transforms.InsertField.type": "org.apache.kafka.connect.transforms.InsertField$Value",
"transforms.InsertField.static.field": "TARGET_SYSTEM",
"transforms.InsertField.static.value": "GOLDENORE CDC"}

4. MASK FIELD

Example:

{"transforms": "Name",
"transforms.Name.type": "org.apache.kafka.connect.transforms.MaskField$Value",
"transforms.Name.fields": "SENDER_NAME",
"transforms.Name.replacement": "MASK SENDER NAME"}

5. TimestampConverter

Dostepne typy danych :

string, unix, Date, Time, or Timestamp.

Example:

{
  "transforms": "StringToNumber,TimestampConverter",
  
  "transforms.StringToNumber.type": "org.apache.kafka.connect.transforms.Cast$Value",
  "transforms.StringToNumber.spec": "epoch:int64",
  
  "transforms.TimestampConverter.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
  "transforms.TimestampConverter.field": "epoch",
  "transforms.TimestampConverter.target.type": "string",
  "transforms.TimestampConverter.format": "yyyy-MM-dd HH:mm:ss.SSS"
}

Rename Field

{"transforms": "RenameField",
"transforms.RenameField.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
"transforms.RenameField.renames": "epoch:time"}

{"transforms": "TimestampConverter",
"transforms.TimestampConverter.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
"transforms.TimestampConverter.format": "yyyy-MM-dd HH:mm:ss",
"transforms.TimestampConverter.field": "epoch",
"transforms.TimestampConverter.target.type": "string"}

6. GDPR Email

Tabela z dostępnymi ustawieniami

Opis	Wzór	Wartość domyślna
Nazwa pola	"transforms.<nazwa transformaty>.fields": "<nazwa kolumny w której znajduje się mail>"	BRAK
Pole określające czy chcemy aby dane generowane przez system były zbliżone do ciągu wejściowego ( email format )	"transforms.<nazwa transformaty>.email.format": <true lub false>	true
Pole określające czy w przypadku nieprawidłowego formatu email ma być wstawiany placeholder	transforms.<nazwa transformaty>.want.placeholder": <true lub false>	true
Pole określające klucz dla szyfrowania AES	"transforms.<nazwa transformaty>.key": "<klucz AES>"	"770A8B65BB156D24DE2A093277530142"
Pole określające czy nie zważając na format danych chcemy zaszyfrować dane algorytmem AES	"transforms.<nazwa transformaty>.encryption": <true lub false>	false
Pole określające czy nie zważając na format danych chcemy deszyfrować dane algorytmem AES	"transforms.<nazwa transformaty>.decryption": <true lub false>	false

Uwagi dodatkowe

• Pól które posiadają wartości domyślne nie musimy zawierać w definicji transformaty.

• W przypadku braku chęci wstawiania placeholdera dane szyfrowane są za pomocą algorytmu MD5

• W przypadku niemożliwości deszyfrowania (ze względu na niedozwolone znaki w kolumnie typu kropki) wykonywane jest domyślna operacja dla całej transformaty a więc szyfrowanie słownikowe

Przykładowe transformaty

Transformata z formatem mail(szyfrowanie słownikowe)

{ 
  "transforms": "FieldMailReal",
  "transforms.FieldMailReal.type": "GdprEmail$Value",
  "transforms.FieldMailReal.fields": "email_adress",
  "transforms.FieldMailReal.email.format": true
}

lub

{
  "transforms": "FieldMail",
  "transforms.FieldMail.type": "GdprEmail$Value",
  "transforms.FieldMail.fields": "email_adress"
}

Transformata bez formatu mail(szyfrowanie SHA256) z placeholderem w przypadku, gdy dana w tabeli nie jest mailem

{
  "transforms": "FieldMailNotReal",
  "transforms.FieldMailNotReal.type": "GdprEmail$Value",
  "transforms.FieldMailNotReal.fields": "email_adress",
  "transforms.FieldMailNotReal.email.format": false,
  "transforms.FieldMailNotReal.want.placeholder": true 
}

lub

{
  "transforms": "FieldMailNotReal",
  "transforms.FieldMailNotReal.type": "GdprEmail$Value",
  "transforms.FieldMailNotReal.fields": "email_adress",
  "transforms.FieldMailNotReal.email.format": false 
}

Transformata bez formatu mail(szyfrowanie SHA256) z szyfrowaniem MD5 w przypadku, gdy dana w tabeli nie jest mailem

{
  "transforms": "FieldMailNotReal",
  "transforms.FieldMailNotReal.type": "GdprEmail$Value",
  "transforms.FieldMailNotReal.fields": "email_adress",
  "transforms.FieldMailNotReal.email.format": false,
  "transforms.FieldMailNotReal.want.placeholder": false 
}

Transformata szyfrująca AES

{
  "transforms": "FieldEncryption",
  "transforms.FieldEncryption.type": "GdprEmail$Value",
  "transforms.FieldEncryption.fields": "email_adress",
  "transforms.FieldEncryption.encryption": true 
}

lub

{
  "transforms": "FieldEncryption",
  "transforms.FieldEncryption.type": "GdprEmail$Value",
  "transforms.FieldEncryption.fields": "email_adress",
  "transforms.FieldEncryption.key": "770A8B65DB156D24ED2C093277530142",
  "transforms.FieldEncryption.encryption": true 
}

Transformata deszyfrująca AES

{
  "transforms": "FieldDecryption",
  "transforms.FieldDecryption.type": "GdprEmail$Value",
  "transforms.FieldDecryption.fields": "email",
  "transforms.FieldDecryption.decryption": true 
}

lub

{
  "transforms": "FieldDecryption",
  "transforms.FieldDecryption.type": "GdprEmail$Value",
  "transforms.FieldDecryption.fields": "email",
  "transforms.FieldDecryption.key": "770A8B65DB156D24ED2C093277530142",
  "transforms.FieldDecryption.decryption": true 
}

7. GDPR Pesel

Tabela z dostępnymi ustawieniami

Opis	Wzór	Wartość domyślna
Nazwa pola	"transforms.<nazwa transformaty>.fields": "<nazwa kolumny w której znajduje się pesel>"	BRAK

Uwagi dodatkowe

• W przypadku, gdy w podanym polu wartość nie jest numerem pesel wstawiana zostaje wartość 0.

Przykładowa transformata

{
  "transforms": "FieldPesel", 
  "transforms.FieldPesel.type": "GdprPesel$Value", 
  "transforms.FieldPesel.fields": "pesel" 
}

8. GDPR Phone

Tabela z dostępnymi ustawieniami

Opis	Wzór	Wartość domyślna
Nazwa pola	"transforms.<nazwa transformaty>.fields": "<nazwa kolumny w której znajduje się numer telefonu>"	BRAK

Uwagi dodatkowe

• W przypadku, gdy w podanym polu wartość nie jest numerem telefonu wstawiana zostaje wartość 48 z następstwem zer w zależności od długości ciągu wejściowego. W przypadku podania wartości krótszej niż 3 cyfry wstawiana jest wartość 0.

• Obsługiwane są numery międzynarodowe.

• Nie obsługiwane są numery wewnętrzne, alarmowe oraz operatora.

• Numery telefonów mogą posiadać spację między cyframi oraz numer kierunkowy kraju poprzedzony +

Przykładowa transformata

{
  "transforms": "FieldPhoneNumber",
  "transforms.FieldPhoneNumber.type": "GdprPhoneNumber$Value",
  "transforms.FieldPhoneNumber.fields": "phone_number"
}

9. GDPR GdprPassportId

Tabela z dostępnymi ustawieniami

Opis	Wzór	Wartość domyślna
Nazwa pola	"transforms.<nazwa transformaty>.fields": "<nazwa kolumny w której znajduje się numer paszportu lub dowodu osobistego>"	BRAK
Pole określające czy w przypadku nieprawidłowego formatu paszportu lub dowodu osobistego ma być wstawiany placeholder	"transforms.<nazwa transformaty>.want.placeholder": <true lub false>"	true
Pole określające czy chcemy anonimizować numer paszportu czy dowodu osobistego	"transforms.<nazwa transformaty>.type.settings": <"passport" lub "id">"	null

Uwagi dodatkowe

• Pól które posiadają wartości domyślne nie musimy zawierać w definicji transformaty.

• W przypadku braku chęci wstawiania placeholdera dane szyfrowane są za pomocą algorytmu FPE za pomocą cyfr i wielkich liter.

Przykładowe transformaty

Transformata anonimizacji numeru paszportu z placeholderem w przypadku, gdy dana w tabeli nie jest numerem paszportu

{
  "transforms": "FieldPassport",
  "transforms.FieldPassport.type": "GdprPassportId$Value",
  "transforms.FieldPassport.fields": "passport_number",
  "transforms.FieldPassport.want.placeholder": true,
  "transforms.FieldPassport.type.settings": "passport"
}

lub 

{
  "transforms": "FieldPassport",
  "transforms.FieldPassport.type": "GdprPassportId$Value",
  "transforms.FieldPassport.fields": "passport_number",
  "transforms.FieldPassport.type.settings": "passport"
}

Transformata anonimizacji numeru paszportu z szyfrowaniem FPE za pomocą cyfr i wielkich liter w przypadku, gdy dana w tabeli nie jest numerem paszportu

{
  "transforms": "FieldPassport",
  "transforms.FieldPassport.type": "GdprPassportId$Value",
  "transforms.FieldPassport.fields": "passport_number",
  "transforms.FieldPassport.want.placeholder": false,
  "transforms.FieldPassport.type.settings": "passport"
}

Transformata anonimizacji numeru dowodu osobistego z placeholderem w przypadku, gdy dana w tabeli nie jest numerem dowodu osobistego

{
  "transforms": "FieldIdNumber",
  "transforms.FieldIdNumber.type": "GdprPassportId$Value",
  "transforms.FieldIdNumber.fields": "id_number",
  "transforms.FieldIdNumber.want.placeholder": true,
  "transforms.FieldIdNumber.type.settings": "id"
}

lub

{
  "transforms": "FieldIdNumber",
  "transforms.FieldIdNumber.type": "GdprPassportId$Value",
  "transforms.FieldIdNumber.fields": "id_number",
  "transforms.FieldIdNumber.type.settings": "id"
}

Transformata anonimizacji numeru dowodu osobistego z szyfrowaniem FPE za pomocą cyfr i wielkich liter w przypadku, gdy dana w tabeli nie jest numerem dowodu osobistego

{
  "transforms": "FieldIdNumber",
  "transforms.FieldIdNumber.type": "GdprPassportId$Value",
  "transforms.FieldIdNumber.fields": "id_number",
  "transforms.FieldIdNumber.want.placeholder": false,
  "transforms.FieldIdNumber.type.settings": "id"
}

10. GDPR CreditCard

Tabela z dostępnymi ustawieniami

Opis	Wzór	Wartość domyślna
Nazwa pola	"transforms.<nazwa transformaty>.fields": "<nazwa kolumny w której znajduje się numer karty bankowej>"	BRAK

Uwagi dodatkowe

• W przypadku, gdy w podanym polu wartość nie jest numerem pesel wstawiane zostaje 16 zer (w przypadku przechowywania wartości w zmiennej np. bigint wstawiana jest wartość 0).

Przykładowa transformata

{
  "transforms": "FieldCard",
  "transforms.FieldCard.type": "GdprCardNumber$Value",
  "transforms.FieldCard.fields": "credit_card"
}

11. Filter (Confluent)

Tabela z dostępnymi ustawieniami

Opis	Wzór	Wartość domyślna
Określa kryteria używane do filtrowania rekordów, które mają zostać uwzględnione lub wykluczone przez tę transformatę. Notacja używana do zapisu warunku filtrowania to predykat JSON Path, którego opis znajduje się w linku: https://github.com/json-path/JsonPath.	"transforms.<nazwa transformaty>.filter.condition": "<warunek filtrowania>"	BRAK
Określa akcję wykonywaną na rekordach pasujących do predykatu filter.condition. Dopuszczalne wartości to: [include, exclude]. Wartość "include", zostawia rekordy pasujące do predykatu i usuwa wszystkie rekordy, które nie spełniają predykatu. Wartość "exclude" usuwa wszystkie rekordy pasujące do predykatu.	"transforms.<nazwa transformaty>.filter.type": "<"include" lub "exclude">"	BRAK
Określa zachowanie, gdy rekord nie ma pola (pól) użytych w filter.condition. Użyj "fail", aby rzucić wyjątek i nie wykonać zadania konektora, "include", aby przepuścić rekord, lub "exclude", aby go odrzucić.	"transforms.<nazwa transformaty>.missing.or.null.behavior": "<"fail", "include" lub "exclude">"	fail

Przykładowa transformata

{
  "transforms": "filterPoland",
  "transforms.filterPoland.type": "io.confluent.connect.transforms.Filter$Value",
  "transforms.filterPoland.filter.condition": "$[?(@.nation== 'Poland')]",
  "transforms.filterPoland.filter.type": "include", 
  "transforms.filterPoland.missing.or.null.behavior": "fail"
}

Transformata z predykatem

{
  "transforms": "filterPoland",
  "transforms.filterPoland.predicate": "IsPredicate2",
  "transforms.filterPoland.type": "io.confluent.connect.transforms.Filter$Value",
  "transforms.filterPoland.filter.condition": "$[?(@.nation== 'Poland')]",
  "transforms.filterPoland.filter.type": "include", 
  "transforms.filterPoland.missing.or.null.behavior": "fail",
  "predicates": "IsPredicate2",
  "predicates.IsPredicate2.type": "org.apache.kafka.connect.transforms.predicates.TopicNameMatches",
  "predicates.IsPredicate2.pattern": ".*client"
}

Powyższa transformata przedstawia użycie predykatu, który pozwala nam na filtrowanie konkretnej tabeli lub tabel. Gdy replikujemy cały schemat lub wiele tabel, które mają kolumny o takiej samej nazwie filtr zadziałałby na wszystkie tabele. Użycie predykatu typu TopicNameMatches pozwala na wskazanie jednego lub wielu topicków za pomocą regex i rozwiązuje problem filtrowania wszystkich tabel znajdujących się w naszym connectorze.

12. Przykład użycia wielu transformat

{
  "transforms": ["FirstNameField", "LastNameField", "CompanyField",  "PhoneNumberField", "NationField", "CreditCardField", "PeselField", "EmailField", "PassportField", "PolishIdField"],

 
  "transforms.FirstNameField.type": "org.apache.kafka.connect.transforms.MaskField$Value",
  "transforms.FirstNameField.fields": "first_name",
  "transforms.FirstNameField.replacement": "First Name",

  "transforms.LastNameField.type": "org.apache.kafka.connect.transforms.MaskField$Value",
  "transforms.LastNameField.fields": "last_name",
  "transforms.LastNameField.replacement": "Last Name",

  "transforms.CompanyField.type":"FpeField$Value",
  "transforms.CompanyField.key":"3677397A24432646294A404D63516654",
  "transforms.CompanyField.fields":"company",

  "transforms.PhoneNumberField.type": "GdprPhoneNumber$Value",
  "transforms.PhoneNumberField.fields": "phone_number",

  "transforms.NationField.type":"FpeField$Value",
  "transforms.NationField.key":"3677397A24432646294A404D63516654",
  "transforms.NationField.fields":"nation",

  "transforms.CreditCardField.type": "GdprCardNumber$Value",
  "transforms.CreditCardField.fields": "credit_card",

  "transforms.PeselField.type": "GdprPesel$Value", 
  "transforms.PeselField.fields": "pesel",

  "transforms.EmailField.type": "GdprEmail$Value",
  "transforms.EmailField.fields": "adress_email",

  "transforms.PassportField.type": "GdprPassportId$Value",
  "transforms.PassportField.fields": "passport_number",
  "transforms.PassportField.type.settings": "passport",

  "transforms.PolishIdField.type": "GdprPassportId$Value",
  "transforms.PolishIdField.fields": "polish_id_number",
  "transforms.PolishIdField.type.settings": "id"
}