NumPy 作為 Python 資料科學領域的核心套件,其陣列操作效率是關鍵。本文將深入探討 NumPy 陣列的索引與廣播機制,並結合實際案例說明如何透過這些技巧提升程式碼效能。首先介紹基本索引和切片操作,接著說明布林遮罩和複合索引的應用,最後以粒子模擬器最佳化為例,展現 NumPy 在處理大量資料時的優勢。透過理解這些技術,開發者能更有效地操作 NumPy 陣列,進而提升整體程式碼執行效率。

基本索引

NumPy陣列的索引類似於Python列表的索引。您可以使用方括號[]來存取陣列中的元素。例如:

import numpy as np

# 建立一個2x2的陣列
arr = np.array([[0, 1], [2, 3]])

# 存取陣列中的元素
print(arr[0, 0])  # Output: 0
print(arr[1, 1])  # Output: 3

切片

NumPy陣列也支援切片,允許您存取陣列中的多個元素。例如:

import numpy as np

# 建立一個2x2的陣列
arr = np.array([[0, 1], [2, 3]])

# 切片
print(arr[0, :])  # Output: [0 1]
print(arr[:, 1])  # Output: [1 3]

複合索引

NumPy陣列也支援複合索引,允許您使用多個索引來存取陣列中的元素。例如:

import numpy as np

# 建立一個2x2的陣列
arr = np.array([[0, 1], [2, 3]])

# 複合索引
print(arr[[0, 1], [0, 1]])  # Output: [0 3]

布林遮罩

NumPy陣列也支援布林遮罩,允許您使用布林值來存取陣列中的元素。例如:

import numpy as np

# 建立一個1x6的陣列
arr = np.array([0, 1, 2, 3, 4, 5])

# 建立一個布林遮罩
mask = np.array([True, False, True, False, False, False])

# 使用布林遮罩存取陣列中的元素
print(arr[mask])  # Output: [0 2]

高階索引

NumPy陣列也支援高階索引,允許您使用多個索引和切片來存取陣列中的元素。例如:

import numpy as np

# 建立一個2x2的陣列
arr = np.array([[0, 1], [2, 3]])

# 高階索引
print(arr[:, [0, 1]])  # Output: [[0 1]
                       #          [2 3]]

效能最佳化

NumPy陣列的索引和切片操作相對較快,但如果您需要更高的效能,可以使用numpy.take()函式。例如:

import numpy as np

# 建立一個1x6的陣列
arr = np.array([0, 1, 2, 3, 4, 5])

# 使用numpy.take()函式存取陣列中的元素
print(np.take(arr, [0, 2]))  # Output: [0 2]

這些是NumPy陣列索引和切片的基本概念和範例。透過使用這些功能,您可以快速存取和操作陣列中的元素,並提高您的程式的效能。

NumPy 的陣列索引和廣播

NumPy 的陣列索引和廣播是兩個非常重要的概念,讓我們可以高效地進行陣列操作。

陣列索引

陣列索引是指如何存取陣列中的元素。NumPy 提供了多種索引方法,包括整數索引、布林索引和切片索引。

import numpy as np

# 整數索引
r_i = np.random.rand(100, 2)
idx = np.arange(50)  # 整數索引
print(np.take(r_i, idx, axis=0))  # 使用 np.take 進行索引

# 布林索引
idx = np.ones(100, dtype='bool')  # 布林索引
print(r_i[idx])  # 使用布林索引進行存取

廣播

廣播是 NumPy 的一種強大功能,允許我們對不同形狀的陣列進行運算。廣播的規則如下:

  • 如果兩個陣列的形狀相同,則運算會在元素級別進行。
  • 如果兩個陣列的形狀不同,則 NumPy 會嘗試將陣列廣播到相同的形狀。
# 元素級別運算
A = np.array([[1, 2], [3, 4]])
B = np.array([[5, 6], [7, 8]])
print(A * B)  # 元素級別乘法

# 廣播
A = np.array([[1, 2], [3, 4]])
B = np.array([5, 6])  # shape (2,)
print(A + B)  # 廣播 B 到 shape (2, 2)
圖表翻譯:
  graph LR
    A[NumPy 陣列] --> B[索引]
    B --> C[整數索引]
    B --> D[布林索引]
    B --> E[切片索引]
    A --> F[廣播]
    F --> G[元素級別運算]
    F --> H[廣播到相同形狀]

內容解密:

NumPy 的陣列索引和廣播是兩個非常重要的概念,讓我們可以高效地進行陣列操作。透過瞭解這些概念,我們可以更好地使用 NumPy 進行資料分析和科學計算。

瞭解 NumPy 的廣播規則

NumPy 的廣播規則是一種強大的功能,允許您在不同形狀的陣列之間進行運算。當您嘗試將兩個陣列相乘或進行其他運算時,NumPy 會嘗試使用廣播規則使陣列的形狀相匹配。

基本廣播規則

如果其中一個運算元是標量(例如,數字),它將被應用於陣列的每個元素。例如:

import numpy as np

A = np.array([[1, 2], [3, 4]])
B = 2

result = A * B
print(result)

輸出:

array([[2, 4],
       [6, 8]])

陣列之間的廣播

如果兩個陣列的形狀不匹配,NumPy 會嘗試從最後一個軸開始匹配形狀。例如:

A = np.array([[1, 2], [3, 4]])
B = np.array([5, 6])

result = A * B
print(result)

輸出:

array([[ 5, 12],
       [15, 24]])

在這個例子中,陣列 B 的形狀為 (2,),而陣列 A 的形狀為 (2, 2)。NumPy 會將陣列 B 廣播到 (2, 2) 的形狀,以匹配陣列 A 的形狀。

新增軸

您可以使用 numpy.newaxis 常數在陣列中新增軸。例如:

A = np.array([[1, 2], [3, 4]])
B = np.array([5, 6])

# 新增軸
B = B[:, np.newaxis]

result = A * B
print(result)

輸出:

array([[ 5, 10],
       [15, 20]])

在這個例子中,陣列 B 的形狀為 (2, 1),而陣列 A 的形狀為 (2, 2)。NumPy 會將陣列 B 廣播到 (2, 2) 的形狀,以匹配陣列 A 的形狀。

圖解廣播規則

以下是廣播規則的圖解:

  (3, 2)       (2,)
  --------  --------
  [1, 2]     [5, 6]
  [3, 4]     [5, 6]
  [5, 6]     [5, 6]

在這個例子中,陣列 (3, 2) 的形狀為 (3, 2),而陣列 (2,) 的形狀為 (2,)。NumPy 會將陣列 (2,) 廣播到 (3, 2) 的形狀,以匹配陣列 (3, 2) 的形狀。

數學運算與廣播

NumPy 提供了多種數學運算,包括基本代數、 三角函式、四捨五入和邏輯運算等。這些運算可以應用於陣列的廣播中,從而實作高效的元素級別運算。

廣播原理

廣播是 NumPy 中的一個重要特性,允許不同形狀的陣列之間進行運算。當兩個陣列的形狀不相同時,NumPy 會自動將陣列廣播到相同的形狀,以便進行元素級別的運算。

例如,假設我們有兩個陣列 AB,分別具有形狀 (5, 1, 2)(5, 2)。我們可以使用廣播原理將 B 陣列廣播到 (5, 1, 2) 的形狀,然後進行元素級別的乘法運算。

import numpy as np

A = np.random.rand(5, 10, 2)
B = np.random.rand(5, 2)

result = A * B[:, np.newaxis, :]

外積運算

外積運算是兩個陣列之間的一種運算,結果是兩個陣列所有可能組合的乘積。例如,假設我們有兩個陣列 ab,分別具有元素 [a1, a2, a3][b1, b2, b3]。外積運算的結果是一個矩陣,包含所有可能組合 (i, j) 的乘積。

a = np.array([1, 2, 3])
b = np.array([4, 5, 6])

result = a[:, np.newaxis] * b[np.newaxis, :]

數學函式

NumPy 提供了多種數學函式,包括平方根、正弦、餘弦等。這些函式可以應用於陣列的元素級別運算。

import numpy as np

array = np.array([4, 9, 16])
result = np.sqrt(array)

比較運算

比較運算是用於過濾陣列元素的條件。例如,假設我們有一個陣列 array,包含隨機數字從 0 到 1,我們可以使用比較運算子過濾出大於 0.5 的元素。

import numpy as np

array = np.random.rand(10)
result = array > 0.5

圖表翻譯

以下是外積運算的 Mermaid 圖表:

  graph LR
    A[a1, a2, a3] -->|外積|> B[b1, b2, b3]
    B -->|乘積|> C[a1*b1, a1*b2, a1*b3, a2*b1, a2*b2, a2*b3, a3*b1, a3*b2, a3*b3]

圖表翻譯:

這個圖表展示了外積運算的過程。左邊的陣列 A 包含元素 [a1, a2, a3],右邊的陣列 B 包含元素 [b1, b2, b3]。外積運算將兩個陣列的元素相乘,產生一個新的矩陣,包含所有可能組合的乘積。

陣列操作與索引

在進行資料分析時,能夠高效地操作和索引陣列是非常重要的。NumPy提供了多種方法來實作這些功能。

條件索引

首先,我們可以使用條件陳述式來索引陣列。例如,要從一個陣列中提取所有大於0.5的數字,可以使用以下方法:

import numpy as np

# 建立一個5x3的隨機陣列
a = np.random.rand(5, 3)

# 使用條件陳述式索引陣列
result = a[a > 0.5]

print(result)

這段程式碼會輸出陣列a中所有大於0.5的元素。

陣列求和

NumPy還提供了多種方法來對陣列進行求和操作。例如,可以使用ndarray.sum方法來對陣列的元素進行求和。這個方法可以根據軸來求和,例如可以對第一軸、第二軸或所有元素進行求和。

import numpy as np

# 建立一個5x3的隨機陣列
a = np.random.rand(5, 3)

# 對第一軸(列)進行求和
sum_axis_0 = a.sum(axis=0)

# 對第二軸(行)進行求和
sum_axis_1 = a.sum(axis=1)

# 對所有元素進行求和
sum_all = a.sum()

print("sum_axis_0:", sum_axis_0)
print("sum_axis_1:", sum_axis_1)
print("sum_all:", sum_all)

這段程式碼會輸出陣列a中各軸的求和結果。

圖表解釋

以下是使用Mermaid語法建立的陣列索引和求和過程圖表:

  flowchart TD
    A[陣列建立] --> B[條件索引]
    B --> C[陣列求和]
    C --> D[輸出結果]
    D --> E[圖表解釋]

圖表翻譯:

這個圖表展示了陣列操作和索引的過程。首先,建立一個陣列,然後使用條件陳述式索引陣列,接著對陣列的元素進行求和,最後輸出結果並進行圖表解釋。

內容解密:

在這個例子中,我們使用NumPy的條件索引和求和方法來運算元組。條件索引允許我們從陣列中提取特定元素,求和方法則允許我們對陣列的元素進行求和。這些方法在資料分析中非常重要,因為它們可以幫助我們快速地處理和分析資料。

計算陣列的總和

在進行資料分析時,計算陣列的總和是一個非常重要的步驟。NumPy 提供了多種方式來計算陣列的總和,包括沿著特定軸的總和和全部元素的總和。

沿著軸的總和

當計算陣列的總和時,可以指定沿著哪個軸進行計算。例如,對於一個二維陣列,可以沿著行(axis=1)或列(axis=0)進行總和計算。

import numpy as np

# 建立一個二維陣列
a = np.array([[1, 2, 3], [4, 5, 6], [7, 8, 9]])

# 沿著行(axis=1)進行總和計算
sum_along_rows = a.sum(axis=1)
print(sum_along_rows)

# 沿著列(axis=0)進行總和計算
sum_along_cols = a.sum(axis=0)
print(sum_along_cols)

###全部元素的總和 如果沒有指定軸,NumPy 會將陣列扁平化並計算所有元素的總和。

# 計算全部元素的總和
total_sum = a.sum()
print(total_sum)

計算向量的範數

線上性代數中,向量的範數(norm)是指向量的大小或長度。對於二維向量,範數可以使用以下公式計算:

範數 = sqrt(x^2 + y^2)

給定一個包含 10 個坐標(x, y)的陣列,想要計算每個坐標的範數,可以使用以下步驟:

  1. 對坐標進行平方,得到一個包含 (x^2, y^2) 元素的陣列。
  2. 使用 NumPy 的 sum 函式沿著最後一個軸進行總和計算。
  3. 對結果進行元素-wise 的平方根計算。

最終的表示式可以壓縮成一行程式碼:

import numpy as np

# 建立一個包含 10 個坐標的陣列
r_i = np.random.rand(10, 2)

# 計算範數
norm = np.sqrt((r_i ** 2).sum(axis=1))
print(norm)

這個程式碼會輸出每個坐標的範數。

最佳化粒子模擬器

在這個章節中,我們將使用 NumPy 來最佳化粒子模擬器。根據我們在第一章中進行的效能分析,發現程式中最慢的部分是以下的迴圈,這個迴圈位於 ParticleSimulator.evolve 方法中:

for i in range(nsteps):
    for p in self.particles:
        norm = (p.x**2 + p.y**2)**0.5
        v_x = (-p.y)/norm
        v_y = p.x/norm
        d_x = timestep * p.ang_vel * v_x
        d_y = timestep * p.ang_vel * v_y
        p.x += d_x
        p.y += d_y

您可能已經注意到,迴圈的主體只作用於當前的粒子。如果我們有一個包含粒子位置和角速度的陣列,我們可以使用廣播操作來重寫迴圈。然而,迴圈的步驟依賴於前一步驟,因此不能以這種方式進行平行化。

因此,很自然地將所有陣列座標儲存在一個形狀為 (nparticles, 2) 的陣列中,角速度儲存在一個形狀為 (nparticles,) 的陣列中,其中 nparticles 是粒子的數量。這些陣列分別稱為 r_iang_vel_i

r_i = np.array([[p.x, p.y] for p in self.particles])
ang_vel_i = np.array([p.ang_vel for p in self.particles])

速度方向是垂直於向量 (x, y) 的,之前是這樣定義的:

內容解密:

上述程式碼使用 NumPy 的陣列操作來最佳化粒子模擬器。首先,建立兩個陣列 r_iang_vel_i 分別儲存粒子的位置和角速度。然後,使用 NumPy 的廣播操作來計算速度方向和位移。

norm = np.linalg.norm(r_i, axis=1)
v_x = -r_i[:, 1] / norm
v_y = r_i[:, 0] / norm
d_x = timestep * ang_vel_i * v_x
d_y = timestep * ang_vel_i * v_y
r_i[:, 0] += d_x
r_i[:, 1] += d_y

圖表翻譯:

此圖示為粒子模擬器的最佳化過程:

  flowchart TD
    A[原始迴圈] --> B[建立陣列]
    B --> C[計算速度方向]
    C --> D[計算位移]
    D --> E[更新粒子位置]

此圖表展示了最佳化過程中各個步驟之間的關係。首先,建立儲存粒子位置和角速度的陣列,然後計算速度方向和位移,最後更新粒子位置。

物理模擬器演算法最佳化

在進行物理模擬時,計算每個粒子的位移和速度是非常重要的步驟。為了提高效率,我們可以使用NumPy來加速計算。

計演算法

首先,我們需要計算每個粒子的速度。假設我們有一個包含所有粒子位置的矩陣 r_i,其中每一行代表一個粒子的x和y坐標。則可以使用以下公式計算速度:

v_x = -r_i[:, 1] / norm_i
v_y = r_i[:, 0] / norm_i

其中,norm_i是每個粒子的位置向量的長度,可以使用以下公式計算:

norm_i = ((r_i ** 2).sum(axis=1)) ** 0.5

然後,我們需要計算每個粒子的位移。假設我們有一個包含所有粒子角速度的矩陣 ang_vel_i,以及時間步長 timestep。則可以使用以下公式計算位移:

d_i = timestep * ang_vel_i[:, np.newaxis] * v_i

其中,v_i是每個粒子的速度,可以使用以下公式計算:

v_i = r_i[:, [1, 0]] / norm_i
v_i[:, 0] *= -1

實作

現在,我們可以實作一個使用NumPy的粒子模擬器。以下是實作的程式碼:

import numpy as np

class ParticleSimulator:
    def __init__(self, particles):
        self.particles = particles

    def evolve_numpy(self, timestep):
        r_i = np.array([(p.x, p.y) for p in self.particles])
        ang_vel_i = np.array([p.ang_vel for p in self.particles])

        norm_i = ((r_i ** 2).sum(axis=1)) ** 0.5
        v_i = r_i[:, [1, 0]] / norm_i
        v_i[:, 0] *= -1

        d_i = timestep * ang_vel_i[:, np.newaxis] * v_i
        r_i += d_i

        for i, p in enumerate(self.particles):
            p.x, p.y = r_i[i]

效能比較

我們可以將這個實作與純Python版本進行比較,來評估NumPy的效能優勢。

import time

# 純Python版本
def evolve_python(self, timestep):
    for p in self.particles:
        # ...

# NumPy版本
def evolve_numpy(self, timestep):
    # ...

# 測試
particles = [Particle() for _ in range(1000)]
simulator = ParticleSimulator(particles)

start_time = time.time()
simulator.evolve_python(0.01)
print("純Python版本:", time.time() - start_time)

start_time = time.time()
simulator.evolve_numpy(0.01)
print("NumPy版本:", time.time() - start_time)

結果顯示,NumPy版本的效能遠遠超過純Python版本。這是因為NumPy可以利用向量化運算和平行處理來加速計算。

進化 NumPy: 高效陣列運算

在這個章節中,我們將深入探討 NumPy 的進化過程,特別是在陣列運算方面的最佳化和應用。

時間步長和步數計算

首先,我們需要計算時間步長 (dt) 和步數 (nsteps)。這兩個引數對於模擬的精確度和效率有著重要影響。

def evolve_numpy(self, dt):
    timestep = 0.00001  # 時間步長
    nsteps = int(dt / timestep)  # 步數計算

初始條件設定

接下來,我們需要設定初始條件,包括粒子的位置 (r_i) 和角速度 (ang_vel_i)。

r_i = np.array([[p.x, p.y] for p in self.particles])  # 粒子位置
ang_vel_i = np.array([p.ang_vel for p in self.particles])  # 角速度

時間演化迴圈

現在,我們可以開始時間演化迴圈。在每個時間步中,我們需要更新粒子的位置和速度。

for i in range(nsteps):
    norm_i = np.sqrt((r_i ** 2).sum(axis=1))  # 節點距離計算
    v_i = r_i[:, [1, 0]]  # 速度計算
    v_i[:, 0] *= -1  # 速度反轉
    v_i /= norm_i[:, np.newaxis]  # 速度歸一化
    d_i = timestep * ang_vel_i[:, np.newaxis] * v_i  # 位移計算
    r_i += d_i  # 位置更新

粒子位置更新

最後,我們需要更新粒子的位置。

for i, p in enumerate(self.particles):
    p.x, p.y = r_i[i]  # 粒子位置更新

效能最佳化和 Benchmarking

為了評估這個演算法的效能,我們可以使用 Benchmarking 工具來測量執行時間和記憶體使用量。同時,我們也可以透過平行化和向量化等技術來最佳化演算法的效能。

import timeit

def benchmark_evolve_numpy(self, dt):
    start_time = timeit.default_timer()
    self.evolve_numpy(dt)
    end_time = timeit.default_timer()
    print(f"Execution time: {end_time - start_time} seconds")

圖表翻譯:

  flowchart TD
    A[初始條件設定] --> B[時間演化迴圈]
    B --> C[粒子位置更新]
    C --> D[效能最佳化和 Benchmarking]
    D --> E[輸出結果]

在這個圖表中,我們可以看到演算法的流程,從初始條件設定到時間演化迴圈、粒子位置更新,最後到效能最佳化和 Benchmarking。這個圖表可以幫助我們更好地理解演算法的流程和最佳化方向。

最佳化粒子模擬器的效能

在上一節中,我們探討瞭如何使用 NumPy 來最佳化粒子模擬器的效能。雖然我們已經看到了一些改善,但提升並不明顯。這是因為 NumPy 的力量在於處理大型陣列時才能體現出來。讓我們增加粒子的數量,看看是否能夠看到更明顯的效能提升。

從底層實作到高階應用的全面檢視顯示,NumPy 陣列的索引與切片機制以及廣播功能,為 Python 的科學計算提供了強大的效能基本。透過多維比較分析,NumPy 在處理大量數值資料時,相較於傳統 Python 列表,展現了顯著的效能優勢,尤其在向量化運算和廣播機制加持下,更能大幅簡化程式碼並提升執行效率。然而,廣播機制也存在一些限制,例如在處理形狀差異過大的陣列時,可能需要額外的形狀調整,並需注意記憶體消耗。技術堆疊的各層級協同運作中體現,NumPy 與其他科學計算函式庫(如 SciPy、Pandas)的無縫整合,構建了更完善的資料科學生態系統。對於追求極致效能的應用,可以進一步探索 NumPy 的底層最佳化技巧,例如使用 np.take() 函式、記憶體對映以及 Cython 等技術。玄貓認為,NumPy 已成為 Python 科學計算不可或缺的基本,深入理解其索引、切片和廣播機制,對於提升程式碼效能至關重要。隨著資料規模的不斷增長,預見 NumPy 的效能優勢將更加突出,並持續推動 Python 在資料科學領域的發展。