https://linen.dev logo
Join Slack
Powered by
Hello, we trying to figure out if it is possible ...
# troubleshooting
l
Hello, we trying to figure out if it is possible to use the stream_maps capabilities to combine/join two streams. This is our meltano.yml. I highlighted the part that is not working and supposed to map a field of the sream 
preferred_language
to the stream 
contacts
if the values are the same. 
....
environments:
- name: dev
config:
plugins:
extractors:
- name: tap-rest-api-msdk
config:
api_url: ****
auth_method: oauth
grant_type: client_credentials
scope: *****
streams:
- name: preferred_language
path: *****
params:
$select: LogicalName
$expand: OptionSet($select=Options)
records_path: $.OptionSet.Options[*]
schema: extract/crm/schema/preferred_language.json
- name: contacts
path: contacts
params:
$top: 10
records_path: $.value[*]
schema: extract/crm/schema/contacts.json
stream_maps:
contacts:
 
gendercode: "'male' if gendercode == 1 else 'female'"
 _*`td_preferred_language: |`*_ 
next(
 
(
 _*`lang['Label_UserLocalizedLabel_Label']`*_ _*`for lang in stream['preferred_language']`*_ _*`if lang['Value'] == td_preferredlanguage`*_ 
),
 
None
 
)
 
....
For example: if stream 
preferred_language
has the following record: 
{"type":"RECORD","stream":"preferred_language","record":{_*"Value":123456,"Label_UserLocalizedLabel_Label":"Danish"*_},"time_extracted":"2024-12-02T13:53:08.711908+00:00"}
and stream 
contacts
has the record: 
{"type":"RECORD","stream":"contacts","record":{"lastname":"Brouwers","firstname":"Peter!","_*td_preferredlanguage":123456*_,"contactid":"0000-0000-0000","telephone1":null,"jobtitle":null,"department":null,"emailaddress1":null,"td_productrolecontact":null,"gendercode":"female","pix_initials":null,"middlename":null},"time_extracted":"2024-12-02T13:53:08.381522+00:00"}
, then we would like to have 
Danish
as new value for stream 
contacts
and field 
td_preferred_language
.
r
I don't think this is possible with stream maps. Sounds like something you would want to use dbt for though.
🙏 1
e
Yeah, Reuben is correct and it's arguably out of scope for stream maps to be able to join streams like this.
💯 1
✅ 1
l
Ok, thanks for clarification!
@Edgar Ramírez (Arch.dev) We have a follow-up question. We have a python lookup table for translating some of our fields of our stream and we would like to do this with a custom mapper. we have the following 
meltano..yml
: 
version: 1
default_environment: dev
project_id: 99290a45-4a7e-42ae-8208-0b2ca725d652
environments:
- name: dev
config:
plugins:
extractors:
- name: tap-rest-api-msdk
config:
api_url:*****
auth_method: oauth
grant_type: client_credentials
scope: *****
streams:
- name: contacts
path: contacts
params:
$top: 10
records_path: $.value[*]
schema: extract/crm/schema/contacts.json
loaders:
- name: target-mssql
config:
database: LiftStaging
default_target_schema: crm
host: localhost # mssql-server
port: '1434'
username: sa
add_record_metadata: false
- name: staging
- name: prod
plugins:
extractors:
- name: tap-rest-api-msdk
variant: widen
pip_url: tap-rest-api-msdk
loaders:
- name: target-mssql
variant: meltanolabs
pip_url: git+<https://github.com/MeltanoLabs/target-mssql.git>
in file 
custom_mappers/translation_mapper.py
we have the following code: 
import json
import sys
def custom_mapper(record):
# Example inline lookup logic
lookup_table = {
907910003: 'french',
}
# Assuming the record has a field 'td_preferredlanguage' that needs to be mapped
if 'td_preferredlanguage' in record:
language_code = record['td_preferredlanguage']
record['language'] = lookup_table.get(language_code, 'Unknown')
return record
we would like to use this custom code as a mapper, to change some field values of a stream, how would we do that? OR is this a wrong approach. If so, what would be a better idea? We don't want to transform data after loading. We just need to do this simple translations. Thank you in advance!!
5 Views
Open in Slack
PreviousNext
Powered by