我正在开发一个项目，该项目从 AWS S3 获取一组输入数据，对其进行预处理和分割，启动 10K 批处理容器以在 AWS Batch 上并行处理分割数据，对数据进行后聚合并推送它到 S3。
我已经从其他项目中获得了 Airflow + Batch 的软件模式，但还没有处理 10k 并行任务的缩放因子。Airflow 很好，因为我可以查看哪些任务失败并在调试后重试任务。但是在一个 Airflow EC2 实例上处理这么多任务似乎是一个障碍。另一种选择是让一项任务启动 10k 容器并从那里监控它。
我没有使用 Step Functions 的经验，但听说它是 AWS 的 Airflow。Step Functions + Batch 在线看起来有很多模式。Step Functions 似乎是检查我的用例的好方法吗？您对失败的工作/重试任务的能力是否与使用 Airflow 获得相同的见解？
我正在尝试加入两个数据框。
我创建了别名并根据这篇文章引用了它们：Spark Dataframe区分具有重复名称的列
但当它命中函数时，我仍然收到有关不明确列的错误fillna()，即使我没有引用有关RetailUnits 的任何内容。
alloc_ns = allocation_num_spots.alias('alloc_ns')
avails_ns = avails_num_spots.alias('avails_ns')
compare_num_avails_inv = avails_ns.join(
alloc_ns,
(F.col('avails_ns.BreakDateTime') == F.col('alloc_ns.AllocationDateTime')) &
(F.col('avails_ns.RetailUnit') == F.col('alloc_ns.RetailUnit')),
how='left').fillna(
{'allocs_sum': 0}).withColumn(
'diff', F.col('avails_sum') - F.col('allocs_sum'))
但我不断收到此错误：
"Reference 'RetailUnit' is ambiguous, could be: avails_ns.RetailUnit, alloc_ns.RetailUnit.;"
Traceback (most recent call last):
File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/dataframe.py", line 1664, in fillna
return DataFrame(self._jdf.na().fill(value), self.sql_ctx)
File "/usr/lib/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py", line 1257, in __call__
answer, self.gateway_client, self.target_id, self.name)
File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/utils.py", line 69, in deco
raise AnalysisException(s.split(': ', 1)[1], stackTrace)
pyspark.sql.utils.AnalysisException: "Reference …假设我有这样的路径：
/folderA/folderB/folderC/item1
/folderA/folderB/folderX/item2
/folderA/folderB/folderF/item1
/folderA/folderB/folderQ/item5
如何获取文件路径的最后两个元素？
folderC/item1
folderX/item2
folderF/item1
folderQ/item5
我知道我可以 git 目录或文件
>>> os.path.basename('/folderA/folderB/folderC/item1')
>>> item1
>>> os.path.dirname('/folderA/folderB/folderC/item1')
>>> /folderA/folderB/folderC/
我有一个类似于下面的数据框。我最初用 -1 填充所有空值以在 Pyspark 中进行连接。
df = pd.DataFrame({'Number': ['1', '2', '-1', '-1'],
'Letter': ['A', '-1', 'B', 'A'],
'Value': [30, 30, 30, -1]})
pyspark_df = spark.createDataFrame(df)
+------+------+-----+
|Number|Letter|Value|
+------+------+-----+
| 1| A| 30|
| 2| -1| 30|
| -1| B| 30|
| -1| A| -1|
+------+------+-----+
处理完数据集后，我需要将所有 -1 替换回空值。
+------+------+-----+
|Number|Letter|Value|
+------+------+-----+
| 1| A| 30|
| 2| null| 30|
| null| B| 30|
| null| A| null|
+------+------+-----+
什么是最简单的方法来做到这一点？
我有一个简单的 dockerfile，用于在 AWS 上运行容器，但在安装时遇到了问题s3fs，这很奇怪，因为我在之前的 dockerfile 中使用了此代码段，没有出现问题。
与分布有关吗？
错误：
multidict/_multidict.c:1:10: fatal error: Python.h: No such file or directory
Dockerfile：
FROM amazonlinux:latest
RUN yum -y install which unzip aws-cli \
&& yum install -y python3-pip python3 python3-setuptools \
&& yum install -y tar.x86_64 \
&& DEBIAN_FRONTEND=noninteractive yum install -y ksh
RUN pip3 install boto3 \
&& yum install -y gcc \
&& pip3 install s3fs
这是输出日志：
Installing collected packages: fsspec, docutils, botocore, typing-extensions, aioitertools, wrapt, attrs, chardet, multidict, async-timeout, idna, …我有一个 python 脚本，可以在使用 AWS SNS 失败时发送电子邮件。我遇到数据帧被截断的问题，并且我不确定如何显示整个线程。我尝试过使用 pandasset_option但似乎没有任何效果。
有没有办法打印出整个数据框并显示整行？
这是代码的一小段：
import pandas as pd
import boto3
pd.set_option('display.max_columns', None)
pd.set_option('display.max_rows', None)
session = boto3.Session()
sns = session.client('sns', region_name='us-east-1')
d = {'failed_runs':['296e62fb-1bc2-4e27-8d69-4a2267237698 - FAILED-GSN-INTO', '92b15078-1fcc-41f0-bc20-d933becf23bf - FAILED-ESP2-INTO', 'b04460ef-32a7-403a-8fbe-468effed7f2b - FAILED-HLN-INTO']}
df = pd.DataFrame(d)
response = sns.publish(TopicArn="arn:aws:sns:xxxxxxxxx",
Message="""
Daily Digest for Failed Runs
------------------------------------------------------------------------------------
{}
------------------------------------------------------------------------------------
""".format(df))
这给了我一封类似的电子邮件：
Daily Digest for Failed Runs
------------------------------------------------------------------------------------
failed_runs
0 296e62fb-1bc2-4e27-8d69-4a2267237698 - FAILED-...
1 92b15078-1fcc-41f0-bc20-d933becf23bf - FAILED-...
2 b04460ef-32a7-403a-8fbe-468effed7f2b - FAILED-...
有没有办法将交叉影线（交叉线）指定为nan_fill_color？我正在创建一些灰度可视化，目前使用白色或黑色作为填充颜色并不能完全传达我想要的含义。
这是我的代码：
state_geo = 'us-states.json'
state_unemployment = 'myData.csv'
state_data = pd.read_csv(state_unemployment)
m = folium.Map(location=[39, -98], zoom_start=4)
folium.Choropleth(
geo_data=state_geo,
name='choropleth',
data=state_data,
columns=['State', 'unemployment'],
key_on='feature.id',
fill_color='Greys',
fill_opacity=1,
line_opacity=.1,
line_weight = 2,
nan_fill_color='Black',
legend_name='Year'
).add_to(m)
folium.LayerControl().add_to(m)
m
链接到 us-states.json 数据：
https://raw.githubusercontent.com/python-visualization/folium/master/examples/data/us-states.json
我有一个日期，它是以下格式的字符串：
'2021-01-16'
并且需要将其转换为以下格式的字符串：
'16-JAN-2021'
我能够像这样得到大部分内容：
x = datetime.strptime('2021-01-16', '%Y-%m-%d')
x.strftime('%d-%b-%Y')
但月份并未完全大写：
'16-Jan-2021'
我有一个像这样的字典：
d = {'cat': 'one', 'dog': 'two', 'fish': 'three'}
给定一个列表，我可以只保留给定的键值吗？
输入：
l = ['one', 'three']
输出：
new_d = {'cat': 'one', 'fish': 'three'}
我是 PostgreSql 的新手，正在研究一个用于学习目的的示例数据库。我有一个面包店数据库和一个食谱表和一个配料表。我试图了解连接两个表的模式，以便配方表有一个引用成分表的成分列表，但我不确定我是否需要第三个表，或者我是否可以只使用这两个表.
CREATE TABLE ingredients
(
ing_id SERIAL PRIMARY KEY,
name varchar(255) NOT NULL,
quantity integer NOT NULL
);
CREATE TABLE recipes
(
rec_id SERIAL PRIMARY KEY,
name varchar(120) NOT NULL,
list_of_ingredients text NOT NULL,
);
编辑：所以假设我在成分表中有这个：
(1, flour, 40)
(2, eggs, 12)
(3, sugar, 23)
这在食谱表中：
(1, cake, "3 flour, 4 eggs, 2 sugar")
我对如何链接这两个表有点困惑。
list_of_ingredients 需要引用成分表作为外键。我知道 NoSQL DB 的全部意义在于允许列表，所以我不确定我是否完全错误地处理了这个问题。
我还将编写一个 Make_Recipe 函数，该函数将接收一个配方并确保有足够的成分，如果通过上述条件，则继续并减少成分数量。
我已经通读了这些帖子，但它们不太符合要求： 用于存储食物食谱的 数据库设计食谱/成分/测量/数量的数据库架构
谢谢你的时间！任何帮助深表感谢。
python ×4
pyspark ×2
airflow ×1
amazon-sns ×1
apache-spark ×1
choropleth ×1
docker ×1
dockerfile ×1
folium ×1
pandas ×1
postgresql ×1
python-3.x ×1
sql ×1