小编New*_*ler的帖子

目的是创建一个大数据框架,我可以在其上执行操作,例如平均每列的行等.

问题是随着数据帧的增加,每次迭代的速度也会增加,所以我无法完成计算.

注意:我df只有两列,这col1是不必要的,所以我加入它.col1是一个字符串,col2是一个浮点数.行数是3k.以下是一个例子:

folder_paths    float
folder/Path     1.12630137
folder/Path2    1.067517426
folder/Path3    1.06443264
folder/Path4    1.049119625
folder/Path5    1.039635769

问题关于如何提高代码效率以及瓶颈在哪里的任何想法？此外,我不确定是否merge可行.

当前的想法我正在考虑的一个解决方案是分配内存并指定列类型:col1是一个字符串,col2是一个浮点数.

df = pd.DataFrame() # create an empty data frame

for i in range(1000):
    if i is 0:
        df = generate_new_df(arg1, arg2)
    else:
        df = pd.merge(df, generate_new_df(arg1, arg2), on='col1', how='outer')

我也尝试过使用pd.concat,但结果非常相似:每次迭代后的时间增加

df = pd.concat([df, get_os_is_from_folder(pnlList, sampleSize, randomState)], axis=1)

pd.concat的结果

run 1
time 0.34s …

我正在尝试使用以下代码创建一个用于代码测试目的pandas DataFrame的hypothesis库：

from hypothesis.extra.pandas import columns, data_frames
from hypothesis.extra.numpy import datetime64_dtypes

@given(data_frames(index=datetime64_dtypes(max_period='Y', min_period='s'),
                   columns=columns("A B C".split(), dtype=int)))

我收到的错误如下：

E           TypeError: 'numpy.dtype' object is not iterable

我怀疑这是因为当我构造DataFramefor时，index=我只传递一个datetime元素，而不是一个ps.Series带有类型的all datetime。即使是这种情况（我不确定），我仍然不确定如何使用hypothesis库来实现我的目标。

谁能告诉我代码有什么问题以及解决方案是什么？

pd.DataFrame如果我有一个这样的列表：

df = pd.DataFrame(np.random.rand(4,5), columns = list('abcde'))
df_list = [df, df]

问题： 如何将其转换为np.array( shape2, 3, 5) 的 3D？

我尝试这样做np.array(df_lsit)，但出现以下错误：

ValueError: cannot copy sequence with size 4 to array axis with dimension 5

Airflow unittest.cfg在AIRFLOW_HOME环境变量路径中创建一个文件。

我的问题是：unittest.cfg为什么我要airflow.cfg通过环境变量同样指向为什么AIRFLOW_CONFIG？

我要这样做的原因是因为我不想在AIRFLOW_HOME目录中有任何配置文件。

另外，如果有人更了解，您能解释一下它的unittest.cfg用途是什么，因为我找不到相关文档。

我有一个里面foo有while True循环的 python 函数。 对于背景：预计会从网络上传输信息，进行一些写作并无限期地运行。断言测试写入是否正确完成。

显然我需要它在某个时候停止，以便进行测试。

我所做的是通过运行multirpocessing并在那里引入超时，但是当我看到测试覆盖率时，通过多处理运行的函数没有被标记为覆盖。

问题 1：为什么 pytest 现在以这种方式工作？

问题 2：我怎样才能做到这一点？

我在想这可能是因为我在技术上退出了循环，所以也许 pytest 没有将其标记为已测试....

import time
import multiprocessing

def test_a_while_loop():
    # Start through multiprocessing in order to have a timeout.
    p = multiprocessing.Process(
        target=foo
        name="Foo",
    )
    try:
        p.start()
        # my timeout
        time.sleep(10)
        p.terminate()
    finally:
        # Cleanup.
        p.join()

    # Asserts below
    ...

更多信息

1. 我考虑添加一个装饰器，例如@pytest.mark.timeout(5)，但这不起作用并且它停止了整个功能，所以我永远不会到达asserts. （如建议here）。
2. 如果我找不到方法，我只会测试零件，但理想情况下，我想找到一种通过打破循环来进行测试的方法。
3. 我知道我可以重写我的代码以使其具有超时，但这意味着更改代码以使其可测试，我认为这不是一个好的设计。
   4. 模拟我没有尝试过（如这里建议的那样），因为我不相信我可以模拟我所做的，因为它从网络上写入信息。我需要真正看到“原始”工作。

我有一个.gitlab-ci.yml看起来像这样的：

image: "python:3.7"

before_script:
  - pip install -r requirements.txt

stages:
  - stageA
  - stageB

stage_a:
  stage: stageA
  script:
  - run_some_python_scripts

stage_b:
  stage: stageB
  script:
  - run_more_python_scripts

通过此设置，requirements.txt在每个阶段之前安装。我只需要安装一次，这样 和 都stageA可以stageB使用。

我怎样才能实现这个目标？

下面我有一个代码,其中读取一个csv文件并700从文件中随机取样.我需要在多个文件上执行此操作,但是如果我遍历文件,那么每个文件的样本(因为它是随机的)将是不同的,我想在随机生成后保持相同.

df = pd.read_csv(file.csv, delim_whitespace=True)
df_s = df.sample(n=700)

我的想法是获取行号,然后将其传递给下一个文件,但这似乎不是很优雅.

你知道这个问题有什么好的解决方案吗？

澄清

文件长度不同,但文件长度最小值为750.

期望的结果

df1 = pd.read_csv(file1.csv, delim_whitespace=True)
df_s1 = df1.sample(n=700) # choose random sample

df2 = pd.read_csv(file2.csv, delim_whitespace=True)
df_s2 = df2.sample(n=700) # use same random sample as above

我有以下多级数据框:

Year   2016                    2017                 
Quarter  3   4                 1                 2      
Month  Sep   Oct   Nov   Dec   Jan  Feb    Mar   Apr   May   Jun
A      0.16  0.95  0.92  0.45  0.30  0.35  0.95  0.88  0.18  0.10
B      0.88  0.67  0.07  0.70  0.74  0.33  0.77  0.21  0.81  0.85
C      0.79  0.56  0.13  0.19  0.94  0.23  0.72  0.62  0.66  0.93

我想总结几个季度,以便最终结果如下:

Year     2016        2017   
Quarter  3     4     1     2
A        0.16  2.32  1.60  1.16
B        0.88  1.44  1.85  1.86
C        0.79  0.89  1.89  2.21

我尝试使用以下公式:

df= df.groupby('Quarter').transform('sum') …

我刚开始阅读有关 Prefect 的文章（并且有一点使用 Airflow 的经验）。

我的目标是设置一个每天在 Prefect 中运行的任务，并将数据收集到一个文件夹中（我想这肯定是 Prefect 可以帮助我做的）。我的目标也是填充历史数据（就像我及时运行这项工作一样）。

在 Airflow 中有 start_date 的概念，当它在过去设置时，DAG 将从该日期开始运行并在每个时间间隔填充。

例如，如果我有一个任务需要一个日期并返回该日期的数据，例如：

# Pseudo code
def get_values_from_somewhere(date: datetime) -> dict:
    return fetched_values_in_json(date)

在 Prefect 中是否有一种本地方式可以做到这一点？我在这里或文档中的任何地方都找不到这个答案，尽管这里提到了回填。任何帮助/指导都将非常有用。

我试过的：

当我设置schedule为：

from datetime import datetime, timedelta

from prefect.schedules import Schedule

schedule = Schedule(clocks=[IntervalClock(interval=timedelta(hours=24), start_date=datetime(2019, 1, 1))])

然后我做flow.run()我只是得到：

INFO:prefect.My-Task:Waiting for next scheduled run at 2020-09-24T00:00:00+00:00

我期望的是从start_date我提供的开始运行，然后暂停直到它到达当前时间并等待下一个时间表。

如果我有以下数据帧,那么派生如下: df = pd.DataFrame(np.random.randint(0, 10, size=(10, 1)))

    0
0   0
1   2
2   8
3   1
4   0
5   0
6   7
7   0
8   2
9   2

是否有一种有效的方式cumsum具有限制的行,每次达到此限制,以启动新的cumsum.达到每个限制(但行数很多)后,将使用总库存创建一行.

下面我创建了一个执行此操作的函数示例,但它非常慢,尤其是当数据框变得非常大时.我不喜欢我的功能是循环,我正在寻找一种方法来使它更快(我猜一种没有循环的方式).

def foo(df, max_value):
    last_value = 0
    storage = []
    for index, row in df.iterrows():
        this_value = np.nansum([row[0], last_value])
        if this_value >= max_value:
            storage.append((index, this_value))
            this_value = 0
        last_value = this_value
    return storage

如果你像我这样朗读我的函数:foo(df, 5) 在上面的上下文中,它返回:

   0
2  10
6  8