You can use a continuous transform for this, however you need 2, one that groups by userId and a 2nd one that merges the output of the 1st transform with topic3 by grouping on externalId.
However for a continuous transform all your source indexes require timestamp field, if you don't have one you might add one using an ingest timestamp. You also need this to connect transform 1 with transform 2.
For picking the right values for the output, e.g. email you can use a script via scripted_metric (spoiler: in a future release this will become easier). Have a look at the provided docs, in the sub sections you will find some painless examples, e.g. the one for top_hits.