如何从Spark中读取/写入dynamodb？
Spark是一个开源的大数据处理框架，而DynamoDB是亚马逊提供的一种高性能、可扩展的NoSQL数据库服务。在Spark中读取/写入DynamoDB可以通过以下步骤实现：
- 首先，确保你已经在Spark环境中安装了相关的依赖库，包括AWS SDK for Java和Spark的AWS Glue库。
- 创建一个SparkSession对象，用于连接Spark和DynamoDB。可以使用以下代码创建一个SparkSession对象：
import org.apache.spark.sql.SparkSession
val spark = SparkSession.builder()
.appName("Spark DynamoDB Example")
.config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
.config("spark.kryo.registrator", "com.amazonaws.services.dynamodbv2.spark.DynamoDBKryoRegistrator")
.getOrCreate()- 使用SparkSession对象创建一个DataFrame，用于读取DynamoDB中的数据。可以使用以下代码读取DynamoDB表中的数据：
val dynamoDBTable = "your-dynamodb-table-name"
val dynamoDBReadOptions = Map(
"tableName" -> dynamoDBTable,
"region" -> "your-dynamodb-region"
)
val dynamoDBDataFrame = spark.read
.format("dynamodb")
.options(dynamoDBReadOptions)
.load()在上述代码中，将"your-dynamodb-table-name"替换为要读取的DynamoDB表的名称，将"your-dynamodb-region"替换为DynamoDB所在的AWS区域。
- 如果需要将Spark DataFrame中的数据写入DynamoDB，可以使用以下代码：
val dynamoDBWriteOptions = Map(
"tableName" -> dynamoDBTable,
"region" -> "your-dynamodb-region"
)
dynamoDBDataFrame.write
.format("dynamodb")
.options(dynamoDBWriteOptions)
.mode("overwrite")
.save()在上述代码中，将"your-dynamodb-table-name"替换为要写入的DynamoDB表的名称，将"your-dynamodb-region"替换为DynamoDB所在的AWS区域。
需要注意的是，以上代码中的读取和写入操作都需要提供正确的DynamoDB表名和AWS区域信息。另外，还可以根据具体需求设置其他读取和写入的选项，例如筛选条件、数据分区等。
推荐的腾讯云相关产品和产品介绍链接地址：
- 腾讯云数据库 TDSQL：https://cloud.tencent.com/product/tdsql
- 腾讯云云服务器 CVM：https://cloud.tencent.com/product/cvm
- 腾讯云云原生容器服务 TKE：https://cloud.tencent.com/product/tke
- 腾讯云对象存储 COS：https://cloud.tencent.com/product/cos
- 腾讯云区块链服务 TBC：https://cloud.tencent.com/product/tbc
- 腾讯云人工智能 AI：https://cloud.tencent.com/product/ai
- 腾讯云物联网平台 IoT Hub：https://cloud.tencent.com/product/iothub
- 腾讯云移动开发 MSDK：https://cloud.tencent.com/product/msdk
请注意，以上链接仅供参考，具体产品选择应根据实际需求和情况进行评估。
相关·内容
下面是我试图从发电机数据库中读取数据并将其加载到数据帧中的代码。 使用scanamo也可以做到这一点吗？ import org.apache.hadoop.io.Text;
import org.apache.hadoop.dynamodb.DynamoDBItemWritable
import org.apache.hadoop.dynamodb.read.DynamoDBInputFormat
import org.apache.hadoop.dynamodb.write.DynamoDBOutputFormat
import org.apache.hadoop.mapred.JobCo
浏览 25提问于2019-12-17得票数 1
回答已采纳
我希望我的Spark应用程序从DynamoDB读取一个表，做一些事情，然后用DynamoDB编写结果。
将表读入DataFrame
现在，我可以将表从DynamoDB作为hadoopRDD读入Spark，并将其转换为DataFrame。但是，我必须使用正则表达式从AttributeValue中提取值。有没有更好/更优雅的方法？在AWS API中找不到任何东西。
package main.scala.util
import org.apache.spark.sql.SparkSession
import org.apache.spark.SparkContext
import org.apac
浏览 6提问于2017-12-08得票数 9
我正在尝试使用pyspark代码从aws emr上的Dynamodb创建一个外部配置单元表。当我在hive提示符上执行该查询时，它工作得很好，但当我将它作为pyspark作业执行时，它就失败了。代码如下：
from pyspark import SparkContext
from pyspark.sql import SparkSession
import os
spark = SparkSession.builder.enableHiveSupport().getOrCreate()
spark.sql('use ash_data')
spark.sql(
浏览 2提问于2019-05-10得票数 2
我的用例是从Spark应用程序写入DynamoDB。由于我对DynamoDB的写入容量有限，并且不想因为成本问题而增加它，我如何将Spark应用程序限制为以规定的速度写入？
这可以通过将分区减少到1然后执行foreachPartition()来实现吗
我已经启用了自动缩放功能，但不想再增加它了。
请建议其他方法处理此问题。
编辑:当Spark应用程序在多节点EMR集群上运行时，需要实现这一点。
浏览 1提问于2018-05-18得票数 2
1回答
我使用的是一个包含3个节点的AWS DAX集群，节点类型为dax.r4.xlarge。当我从EMR集群运行我的spark应用程序时，它总是从dynamodb表中获取值。即使我在相同的键集上运行相同的应用程序，它也在查询dynamodb表。在DAX集群指标中，我看到0个缓存命中和未命中。
浏览 15提问于2020-02-26得票数 0
回答已采纳
我已经写了一个程序来写项目到DynamoDB表中。现在，我想使用PySpark从DynamoDB表中读取所有项。在Spark中有没有可以做到这一点的库？
浏览 0提问于2016-02-05得票数 8
1回答
我们要在AWS堆栈上建造Lambda架构。缺乏开发知识迫使我们更喜欢AWS管理的解决方案而不是自定义部署。
我们的工作流程：
[Batch layer]
Kinesys Firehouse -> S3 -Glue-> EMR (Spark) -Glue-> S3 views -----+
|===> Serving layer (ECS)
浏览 1提问于2018-10-28得票数 0
我正在使用AWS写信给dynamoDB。我使用AWS控制台从dynamoDB读取数据。但是，我已经看到了一些陈旧的阅读和最新的非记录被更新，当试图把记录拉到几分钟以下。什么是数据拉出的安全时间间隔，以确保读取时有最新的数据可用？30分钟是安全间隔吗？
以下是来自AWS网站:只是想了解最近是最近这里。从DynamoDB表读取数据时，响应可能不会反映最近完成的写操作的结果。响应可能包含一些陈旧的数据
你好啊，精灵
浏览 2提问于2022-03-09得票数 1
我正在尝试确定一种解决方案，使用火花流从HBASE表中读取数据，并将数据写入另一个HBASE表。
我在互联网上找到了许多示例，它们要求创建一个DSTREAM来从HDFS文件和all.But中获取数据。我找不到任何示例来从HBASE表中获取数据。
例如，如果我有一个HBASE表'SAMPLE‘，它的列是'name’和'activeStatus‘。如何使用spark streaming从基于activeStatus列的表SAMPLE中检索数据(新数据？
欢迎使用spark streaming从HBASE表检索数据的任何示例。
致敬，Adarsh K S
浏览 0提问于2018-12-11得票数 0
2回答
我正尝试在amazon EMR实例上运行pyspark以从dynamodb读取数据，我想知道如何在我的代码中设置拆分和工作的数量？
我按照下面两篇文档中的说明使用了下面的代码，该代码当前连接到dynamoDB并读取数据。和
from pyspark.context import SparkContext
sc = SparkContext.getOrCreate()
conf = {"dynamodb.servicename": "dynamodb", "dynamodb.input.tableName":
"Table1&
浏览 2提问于2017-12-03得票数 0
我想在dynamodb中维护两个表。一个是主表，另一个是辅表。主表将包含最新数据。辅助表将存储以前版本的数据。我想在主表和辅表之间进行交换，这样API层就可以访问最近的数据。如何在AWS dynamodb中执行此操作？
浏览 3提问于2019-07-11得票数 0
1回答
我有一个场景，一个公司在亚马逊云上有两个区域，在美国有一个区域，在亚洲有第二个区域。在当前的体系结构中，AWS、DynamoDB和MySQL解决方案被使用并安装在美国地区。持有业务逻辑的亚洲地区的EC2服务器必须访问美国地区的DynamoDB和RDS才能获取或更新数据。
该公司现在希望在亚洲地区安装DynamoDB和MySql，以获得更好的性能，这样亚洲地区的EC2服务器就可以从同一地区获得所需的数据。
现在的主要问题是如何在两个区域之间同步数据，当前的DynamoDB和RDS本质上并不支持多个区域。
在这种情况下有什么最佳做法吗？
浏览 1提问于2014-02-23得票数 3
回答已采纳
1回答
从DynamoDB迁移到RDS
、、、、
我正在考虑将DynamoDB数据库迁移到RDS的不同选项。在关系格式中，数据结构更有意义。
有8张表格，每个表约有100万份文件。我们已经计算出主键/外键之间的映射。
从我在AWS上读到的文档来看，我有一些选择。
AWS数据管道-> S3 ->转换为csv -> AWS数据库迁移服务
自定义程序以csv格式将表写入S3 -> AWS数据库迁移
自定义程序从dynamoDB ->中读取，然后逐个表插入RDS表，直到完成。
是否可以使用AWS数据管道直接从DynamoDB复制到RDS？
有没有其他人有过这种移民的经验？还有其他选择吗？
浏览 0提问于2019-01-03得票数 5
回答已采纳
我在Apache作业中使用AWS来用从DynamoDB提取的数据填充S3表。Spark作业只是使用具有非常强流的单个PutItem(仅用于编写的三个m3.x大型节点)编写数据，并且没有任何重试策略。
DynamoDB docs ，即AWS有退避策略，但最终如果速率太高，ProvisionedThroughputExceededException就可以提高。我的火花工作了三天，只受到DynamoDB thoughput (相等500个单位)的限制，所以我期望比率非常高，队列也非常长，但是我没有任何抛出异常或丢失数据的迹象。
所以，我的问题是-当以很高的速率写信给DynamoDB时，什么时候可以得
浏览 7提问于2017-07-19得票数 0
回答已采纳
1回答
DynamoDB条件检查故障监测
、、、、
我使用为Asp.net会话状态提供程序设置了DynamoDB。
在我的ASP.NET_SessionState表中，DynamoDb中有“条件CheckFailed”监视。
我的问题是:在什么情况下，这些例外发生，我如何减少它们？
浏览 2提问于2015-06-17得票数 2
回答已采纳
1回答
在我的安卓应用程序中，我使用亚马逊DynamoDB。我创建了10个具有读取能力10和写容量5的表，今天我收到了一封来自Amazon的电子邮件。我花了11.36美元。
我不明白自由层的含义。以下是我从亚马逊上读到的：
DynamoDB客户可以获得25 GB的免费存储，以及多达25个写入容量单元和25个持续吞吐量容量单位(足够每月处理多达2亿个请求)和250万个来自DynamoDB流的免费读取请求。
请更清楚地告诉我自由层的含义: 25读和25写容量单位！
浏览 3提问于2015-09-03得票数 6
回答已采纳
1回答
需求：使用Scala从本地机器读取DynamoDB(不是本地的，而是在AWS上)的数据。
当我们使用电子病历集群时，可以使用emr-hadoop-dynamodb.jar读取Understanding:数据。
问题
可以使用emr-dynamodb-hadoop.jar?EMR集群从DynamoDB(云上而不是本地)读取数据，而不是使用集群。我直接想要使用本地机器上的scala代码访问dynamodb
build.sbt
version := "0.1"
scalaVersion := "2.11.12"
scalacOptions := Seq("-
浏览 0提问于2020-11-19得票数 2
1回答
我正在尝试使用spark从亚马逊s3获取一个文件(以DataFrame或RDD的形式)，执行一些简单的转换，然后将文件发送到DynamoDB上的一个表中。
在阅读了其他一些论坛帖子后，我开始了解到读/写DynamoDB需要使用hadoopRDD -这与spark中的RDD不同-也不同于我检索s3文件的方式。
我该如何将DataFrame/RDD从s3中的文件更改为hadoopRDD，这样我才能将其发送回来？
我正在使用scala并测试spark-shell中的所有东西。
再次提前感谢！
浏览 0提问于2016-05-26得票数 7
2回答
我是在将数据从DynamoDb加载到Redshift的上下文中这样问的。根据
为了避免消耗过多的提供的读取吞吐量，我们建议您不要从生产环境中的Amazon表加载数据。
我的数据正在制作中，所以我如何才能把它拿出来呢？
或者，DynamoDB流是将数据从DynamoDB转移到Redshift的更好的总体选择吗？(我知道这不会增加我的RCU成本。)
浏览 6提问于2017-11-13得票数 0
回答已采纳
DynamoDB的冲突解决策略是什么？关于Dynamo的白皮书讨论了如何通过GetItem返回多个版本，由客户端来解决。
因此，说Dynamo和DynamoDB是不同的，GetItem只返回一个值。在这种情况下，DynamoDB采用的冲突解决策略是什么？
浏览 2提问于2017-11-16得票数 2
回答已采纳
我正在尝试学习如何在DynamoDB内部进行写入/更新。这就是我能找到的。
当您的应用程序将数据写入DynamoDB表并收到HTTP200响应(OK)时，数据的所有拷贝都将更新。数据最终将在所有存储位置保持一致，通常在一秒或更短时间内。
例如:如果我的DynamoDB有50个分区，并且在一个区域中跨3个可用区进行复制，那么在DynamoDB中会发生什么
After it receives an API request to create an item
After it sends the 200 OK response to the client
我真的很感激任何文件，谈到这一点或直接从您
浏览 0提问于2017-01-17得票数 0
假设我使用的是最终一致的读取模式。在第一次写入值{"name":"Bob"，"age":"1"}的键后，我的读碰巧从副本(旧值{"name":"Bob"，"age":"0"})返回陈旧的数据，然后再用{"name":"Cat"，"age":"0"}进行第二次写入(基于此陈旧读取)。
那么在DynamoDb侧会发生什么呢？
拒绝使用值{“”：“Cat”、“age”：“0”}更新的值{“名称”：“Cat”、“
浏览 2提问于2020-11-04得票数 0
回答已采纳
比方说，我们有一种情况，不是在表中获取总计数，而是获取具有特定状态的记录的计数。我们知道DynamoDb是无模式的，仍然需要逐条计算每条记录才能得到总计数。但是，我们如何使用dynamoDb查询来利用上述需求呢？
浏览 1提问于2019-10-09得票数 1
我有一个DynamoDB表，需要连接到EMR Spark SQL才能在该表上运行查询。我得到了EMR Spark Cluster，上面有版本标签emr-4.6.0和Spark 1.6.1。
我指的是文档：
连接到主节点后，我运行以下命令：
spark-shell --jars /usr/share/aws/emr/ddb/lib/emr-ddb-hadoop.jar
它给出一个警告：
Warning: Local jar /usr/share/aws/emr/ddb/lib/emr-ddb-hadoop.jar does not exist, skipping.
稍后，当我使用以下命令导入Dy
浏览 3提问于2016-05-06得票数 5
情况：
一个带有API的web服务，用于从DynamoDB读取记录。它使用最终一致的读取(GetItem默认模式)
一个由两个步骤组成的集成测试：
在DynamoDB中创建测试数据
调用服务以验证它是否返回预期的结果。
我担心，由于数据的最终一致性，这个测试肯定是脆弱的。
如果我试图在使用GetItem withConsistenRead=true编写之后立即验证数据，它只保证数据已经写入大多数DB副本，而不是全部，因此，正在测试的服务在下一步仍然有机会从未更新的副本中读取数据。
是否有方法确保数据在继续之前已写入所有 DynamoDB副本？
浏览 2提问于2017-02-10得票数 1
回答已采纳
我想让DynamoDB中的主键成为像SQL一样的AutoIncrement键。在生成后，如果序列中有缺失的项，(在删除一些项之后)
+--------+---------------+
| id | name |
+--------+---------------+
| 1 | AuctionStart |
+--------+---------------+
| 2 | AuctionEnd |
+--------+---------------+
| 5 | Bid |
+--------+----
浏览 20提问于2021-11-19得票数 1
我从其中下载了一个url (它是JSON格式的)，使用Databricks：
url="https://tortuga-prod-eu.s3-eu-west-1.amazonaws.com/%2FNinetyDays/amzf277698d77514b44"
testfile = urllib.request.URLopener()
testfile.retrieve(url, "file.gz")
with gzip.GzipFile("file.gz", 'r') as fin:
json_bytes = fin.read()
浏览 3提问于2022-04-22得票数 0
当表中没有活动时，启用了自动缩放的DynamoDB将不会缩减。这是因为它使用CloudWatch警报，该警报要求对消耗的读或写单元进行活动(取决于您想要的扩展类型)。
那么，当表中没有活动时，如何让DynamoDB缩减规模呢？
浏览 0提问于2017-09-01得票数 3
我的问题是的时间复杂度.
我知道DynamoDB操作是从客户端(例如AWS或EC2中的boto3 )抽象出来的，在DynamoDB本身内--但是不管怎么说，操作的运行时复杂性会影响 ，还是会影响调用DynamoDB操作的计算运行时(例如Lambda / EC2) ？
我还知道(在一般复杂性理论中)从有序的列表或数组中删除一个元素，这样删除的元素之后/之后的列表/数组元素就会向左移动1，以替换在被删除元素的原始索引/位置上留下的空/空空白，这是一个线性O(n)过程:以下所有元素都必须移动，并且它们的数量是O(n)的顺序。
这样做的最终结果(元素转移到弥补差距)，也特别是在每个RE
浏览 1提问于2020-07-04得票数 0
我想在不使用EMR集群的情况下运行本地Dynamodb spark作业，该作业从一些表中读取数据并将其写入到parquet / CSV文件中。我没有找到任何支持这一点的火花发电机连接器，也许你有什么想法？
我的代码示例：
import org.apache.hadoop.dynamodb.DynamoDBItemWritable
import org.apache.hadoop.dynamodb.read.DynamoDBInputFormat
import org.apache.hadoop.io.Text
import org.apache.hadoop.mapred.JobConf
im
浏览 37提问于2018-09-05得票数 0
回答已采纳
我有下面的函数从dynamodb中提取所需的列，它运行得很好。
问题是，它只从表中提取了几行。
表中有26000+行，但是我只能在这里得到3000行。我错过了什么吗？
def get_columns_dynamodb():
try:
response = table.query(
ProjectionExpression= " id, name, date",
KeyConditionExpression=
Key('opco_type').eq('cwc&#
浏览 3提问于2022-01-04得票数 0
我想把我的数据库从dynamodb迁移到redis。我在redis中找不到dynamodb的导出机制和导入机制之间的联系。如果您能回答我要遵循的步骤，我将非常感激。
浏览 27提问于2016-09-18得票数 0
2回答
我需要将数据从一个生成器表复制到另一个生成器表，并在此过程中进行一些转换。为此，我将数据从源表导出到s3，并在其上运行crawler。在我的胶水作业中，我使用了以下代码： mapped = apply_mapping.ApplyMapping.apply(
frame=source_df,
mappings=[
("item.uuid.S", "string", "uuid", "string"),
("item.options.SS", "set"
浏览 18提问于2021-02-12得票数 3
回答已采纳
1回答
嗨，我在AWS glue spark工作。我从dynamodb表中获取数据并从中创建一个动态框架。我希望能够发送该表中的所有数据，记录在sqs中。我看到了另一个将动态帧转换为spark数据帧的建议。但这将是一个包含数百万条记录的表。转换为数据帧可能需要一段时间。我希望能够将动态帧中的所有记录发送到sqs队列。
下面是我的代码：
sqs = boto3.resource('sqs')
sqs_queue_url = f"https://sqs.us-east-1.amazonaws.com/{account_id}/my-stream-queue"
queue
浏览 3提问于2020-09-28得票数 1
2回答
在DynamoDB中压缩现有记录
、、、、
嗨，我面临着400KB的限制与DynamoDB的记录较大的大小。我计划在将记录插入到DynamoDB之前使用Gzip压缩记录，但挑战是如何在DynamoDB中压缩现有记录。有没有人能帮我解决压缩现有记录的问题？
浏览 1提问于2021-04-26得票数 0
2回答
我需要从Source DynamoDB表中完成一些分组工作，然后将每个结果项写入另一个目标DynamoDB表(或源表的辅助索引)。
我看到DynamoDB可以用作源(也可以在中报告)。
但是，我不清楚是否可以将DynamoDB表用作。
注意:必须将每个生成的分组项写入一个单独的DynamoDB项(即，如果分组产生了X个对象，则必须将X项写入目标DynamoDB表)。
浏览 7提问于2020-04-13得票数 5
回答已采纳
我有一组亚马逊网络服务实例，其中Apache Hadoop发行版和apache spark被设置。我试图通过Spark streaming访问DynamoDb来读写表，但在编写Spark- DynamoDB代码的过程中，我了解到需要emr-ddb-hadoop.jar来获取DynamoDB输入格式和只存在于EMR集群中的OutputFormat。在查看了一些博客之后，似乎只有使用EMR Spark才能访问它。这是正确的吗？
但是，我使用独立的JAVA SDK来访问Dynamodb，它工作得很好
浏览 0提问于2016-04-07得票数 2
下面是代码片段：
val client = AmazonDynamoDBClientBuilder.standard.withRegion(Regions.the_region).withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials("access_key", "secret_key"))).build()
val dynamoDB = new DynamoDB(client)
val table = dynamoDB.getTable("tbl_name"
浏览 4提问于2017-08-01得票数 1
我试图以编程的方式将dynamodb表加载到HDFS中(通过java，而不是通过hive)，我无法在网上找到如何做到这一点的示例，所以我想下载包含org.apache.hadoop.hive.dynamodb的jar并逆向工程这个过程。
不幸的是，我也找不到这个文件。
谁能帮我回答以下问题(按优先顺序排列)。
将dynamodb表加载到HDFS中的Java示例(该示例可以作为表输入格式传递给映射程序)。
包含org.apache.hadoop.hive.dynamodb的罐子。
谢谢!
浏览 1提问于2013-06-13得票数 1
我正在创建无服务器框架项目。
DynamoDB表由其他CloudFormation堆栈创建。
如何在serverless.yml中引用现有的dynamodb表的serverless.yml
我的配置如下
resources:
Resources:
MyDbTable: //'arn:aws:dynamodb:us-east-2:xxxx:table/MyTable'
provider:
name: aws
...
onDBUpdate:
handler: handler.onDBUpdate
events:
- stream:
浏览 0提问于2017-12-27得票数 4
我一直在努力理解AWS DynamoDB文档中WCU的含义。我从AWS文档中了解到
如果您的应用程序需要写入每项大小为0.2KB的1000项，则需要提供1000 WCU (即0.2/1 = 0.2，使最近的1KB，所以1000项(要写入)* 1KB() =1000 WCU)。
如果我以上的理解是正确的，那么对于那些需要每秒将数百万条记录写入DynamoDB的应用程序，这些应用程序需要提供数百万WCU吗？
谢谢你能澄清我。
浏览 1提问于2017-12-19得票数 1
回答已采纳
我正在使用以下脚本将数据从DynamoDB导出到S3：
CREATE EXTERNAL TABLE TableDynamoDB(col1 String, col2 String)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler' TBLPROPERTIES (
"dynamodb.table.name" = "TableDynamoDB",
"dynamodb.column.mapping" = "col1:col1,col2:col2&#
浏览 1提问于2012-07-26得票数 3
回答已采纳
我的Kafka主题有三个分区，我想知道我是否可以从三个分区中的一个读取。我的客户是spark structured流媒体应用。 下面是我在spark中现有的kafka设置。 val inputDf = spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers", brokers)
.option("subscribe", topic)
.option("startingOffsets", "latest")
.lo
浏览 15提问于2019-02-15得票数 2
回答已采纳
1回答
目前，我正试图找到一个(很好的)解决方案，说明如何将外部MySql数据库中的数据同步，该数据库与任何AWS完全分离为AWS DynamoDb。
同步。进程应该每天12:00左右运行，并且应该从DynamoDb获取包含创建日期的最新插入项，以确保我们只从同步时的给定日期/时间获取MySql数据。跑啊。同步。通常每天都会传送大约110.000张唱片。
有一点要注意:我们在我工作的地方使用.NET。
据我所知，有几个AWS服务可以帮助我这样做：
ERM ()
AWS似乎是要走的路，但似乎Hive脚本无法与外部MySql数据库通信？还是我搞错了？我发现很难找到任何可用的Hive脚本示例。
数据管道 (
浏览 3提问于2016-08-24得票数 1
1回答
我正在探索从Spark日志中为Spark程序获取数据谱系信息的方法。
我正在寻找像kafka主题或表Spark程序读取或写入的信息，以便我们可以获得该信息运行时，并建立端到端的数据流移动。有没有人探索过这种框架。
当我有Info日志级别设置时，我可以获得有关输入kafka读取和表写入数据的信息，但是，如果数据被发送到Kafka主题或输入表读取，我无法获得信息。
感谢您的帮助。
谢谢&致以问候。
浏览 3提问于2017-11-13得票数 0
我想从星火流应用程序中使用DynamoDB流。
火花流使用KCL从Kinesis读取。有一个库可以让KCL能够从DynamoDB流中读取:kinsis-适配器。
但是，是否有可能将这个库插入火花呢？有人做过这个吗？
我正在使用Spark2.1.0。
我的备份计划是让另一个应用程序从DynamoDB流读取到一个动态流中。
谢谢
浏览 3提问于2017-04-16得票数 5
回答已采纳
1回答
CalcDf().show结果为show本身分为13个阶段(0-12) +1( 13 )。
当我试图将结果写入表时，我假设应该只有13个阶段(0-12)，而不是看到和附加阶段( 13 )。它是从哪里来的，它是干什么的？我不会执行任何需要洗牌的重新分区或其他操作。据我所知，spark应该只将1100个文件写入表中，但这不是正在发生的事情。
CalcDf()
.write
.mode(SaveMode.Overwrite)
.insertInto("tn")
CalcDf()逻辑
val dim = spark.sparkContext.broadcast(
spark
浏览 4提问于2021-07-13得票数 1
回答已采纳
我们正试图为规则引擎提供一个在AWS中实现的设计。下面是对这种情况的解释。
我们正在AWS上构建一个数据湖，使用S3作为存储，具有不同的存储桶来表示数据丰富。例如，数据落入原始桶中。从中选择数据并进行数据标准化(有些标准化使布尔字段统一，如有值0或1，将所有数据格式列转换成一个非格式的标准，如dd/mm/yyyy :mi:ss，对某些列数据的裁剪空间等)。
数据移动使用胶水作业、电火花代码、用于编排的Step函数和用于存储作业配置的Dynamodb以及用于存储元数据的glue目录来完成。
需要在Dynamodb中以这种方式存储标准化规则，这样胶水作业就可以从DynamoDB表中
浏览 0提问于2020-09-07得票数 0
1回答
分布式最终一致性密钥库
、、、
我发现很难说服自己使用像DynamoDB这样的复杂设计，而不是简单的复制策略。
假设我们希望在5个服务器上构建一个分布式密钥/值数据存储。(每个服务器都有完全相同的副本)。
像DynamoDB一样，最终一致性系统通常使用复杂的冲突协调、向量时间戳等来实现最终的一致性。
但相反，为什么我们不能简单地做以下几点：
对于写，客户端将向所有服务器发出写命令。因此，所有服务器都将以相同的顺序执行客户端的写命令。它将在服务器提交写入之前回复客户端。
对于read，客户端将只执行一个循环，每次只有一个服务器负责读取命令。(其他服务器不会看到read命令)是的，客户端可能会经历暂时陈旧的数据，但最终所
浏览 0提问于2017-05-16得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
