Comment utiliser Apache Streaming avec DynamoDB Stream

Question

Nous avons une exigence dans laquelle nous enregistrons les événements dans une table DynamoDB lorsqu'une annonce est exposée à l'utilisateur final. Il y a plus de 250 écritures dans cette table par seconde dans la table dynamoDB.

Nous souhaitons agréger et déplacer ces données vers Redshift pour l'analyse.

Le flux DynamoDB sera appelé pour chaque insertion faite dans la table je suppose. Comment puis-je alimenter le flux DynamoDB en une sorte de lots et ensuite traiter ces lots. Existe-t-il des bonnes pratiques concernant ce type de cas d'utilisation? Je lisais à propos de l'étincelle apache et il me semble qu'avec Apache Spark nous pouvons faire ce genre d'agrégation. Mais le flux d'étincelles Apache ne lit pas le flux DynamoDB.

Toute aide ou pointeur est appréciée.

Merci

Answer 1

flux de DynamoDB ont deux interfaces: API de bas niveau, et l'adaptateur Kinesis. Apache Spark a un Kinesis integration, vous pouvez donc les utiliser ensemble. Dans le cas où vous vous demandez quelle interface de flux DynamoDB vous devriez utiliser, AWS suggère que l'adaptateur Kinesis est recommandé. Pour utiliser Kinesis adapter for DynamoDB

peu plus de choses à considérer:

• Au lieu d'utiliser Apache Spark, il est intéressant de regarder Apache Flink. C'est une solution de premier ordre (Spark implémente le streaming en utilisant le micro-batching), avec des latences plus faibles, un débit plus élevé, des opérateurs de streaming plus puissants et un support pour le traitement des cycles. Il a également un

• Il se peut que vous n'ayez pas besoin de flux DynamoDB pour exporter des données vers Redshift. Vous pouvez exporter des données à l'aide des commandes Redshift.

Answer 2

Amazon EMR fournit une implémentation de ce connecteur dans le cadre de emr-Hadoop-ddb.jar, qui contient la classe DynamoDBItemWriteable. En utilisant cette classe, vous pouvez implémenter votre propre DynamoDBInputFormat comme indiqué ci-dessous.

public class DynamoDbInputFormat implements InputFormat, Serializable { 

    @Override 
    public InputSplit[] getSplits(@NonNull final JobConf job, final int numSplits) throws IOException { 
     final int splits = Integer.parseInt(requireNonNull(job.get(NUMBER_OF_SPLITS), NUMBER_OF_SPLITS 
      + " must be non-null")); 

     return IntStream. 
      range(0, splits). 
      mapToObj(segmentNumber -> new DynamoDbSplit(segmentNumber, splits)). 
      toArray(InputSplit[]::new); 
}